You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1422 lines
44 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. package cos
  2. import (
  3. "context"
  4. "crypto/md5"
  5. "encoding/hex"
  6. "encoding/json"
  7. "encoding/xml"
  8. "errors"
  9. "fmt"
  10. "hash/crc64"
  11. "io"
  12. "io/ioutil"
  13. "net/http"
  14. "net/url"
  15. "os"
  16. "sort"
  17. "strconv"
  18. "strings"
  19. "time"
  20. )
  21. // ObjectService 相关 API
  22. type ObjectService service
  23. // ObjectGetOptions is the option of GetObject
  24. type ObjectGetOptions struct {
  25. ResponseContentType string `url:"response-content-type,omitempty" header:"-"`
  26. ResponseContentLanguage string `url:"response-content-language,omitempty" header:"-"`
  27. ResponseExpires string `url:"response-expires,omitempty" header:"-"`
  28. ResponseCacheControl string `url:"response-cache-control,omitempty" header:"-"`
  29. ResponseContentDisposition string `url:"response-content-disposition,omitempty" header:"-"`
  30. ResponseContentEncoding string `url:"response-content-encoding,omitempty" header:"-"`
  31. Range string `url:"-" header:"Range,omitempty"`
  32. IfModifiedSince string `url:"-" header:"If-Modified-Since,omitempty"`
  33. // SSE-C
  34. XCosSSECustomerAglo string `header:"x-cos-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  35. XCosSSECustomerKey string `header:"x-cos-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  36. XCosSSECustomerKeyMD5 string `header:"x-cos-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  37. //兼容其他自定义头部
  38. XOptionHeader *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  39. XCosTrafficLimit int `header:"x-cos-traffic-limit,omitempty" url:"-" xml:"-"`
  40. // 下载进度, ProgressCompleteEvent不能表示对应API调用成功,API是否调用成功的判断标准为返回err==nil
  41. Listener ProgressListener `header:"-" url:"-" xml:"-"`
  42. }
  43. // presignedURLTestingOptions is the opt of presigned url
  44. type presignedURLTestingOptions struct {
  45. authTime *AuthTime
  46. }
  47. // Get Object 请求可以将一个文件(Object)下载至本地。
  48. // 该操作需要对目标 Object 具有读权限或目标 Object 对所有人都开放了读权限(公有读)。
  49. //
  50. // https://www.qcloud.com/document/product/436/7753
  51. func (s *ObjectService) Get(ctx context.Context, name string, opt *ObjectGetOptions, id ...string) (*Response, error) {
  52. var u string
  53. if len(id) == 1 {
  54. u = fmt.Sprintf("/%s?versionId=%s", encodeURIComponent(name), id[0])
  55. } else if len(id) == 0 {
  56. u = "/" + encodeURIComponent(name)
  57. } else {
  58. return nil, errors.New("wrong params")
  59. }
  60. sendOpt := sendOptions{
  61. baseURL: s.client.BaseURL.BucketURL,
  62. uri: u,
  63. method: http.MethodGet,
  64. optQuery: opt,
  65. optHeader: opt,
  66. disableCloseBody: true,
  67. }
  68. resp, err := s.client.doRetry(ctx, &sendOpt)
  69. if opt != nil && opt.Listener != nil {
  70. if err == nil && resp != nil {
  71. if totalBytes, e := strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64); e == nil {
  72. resp.Body = TeeReader(resp.Body, nil, totalBytes, opt.Listener)
  73. }
  74. }
  75. }
  76. return resp, err
  77. }
  78. // GetToFile download the object to local file
  79. func (s *ObjectService) GetToFile(ctx context.Context, name, localpath string, opt *ObjectGetOptions, id ...string) (*Response, error) {
  80. resp, err := s.Get(ctx, name, opt, id...)
  81. if err != nil {
  82. return resp, err
  83. }
  84. defer resp.Body.Close()
  85. // If file exist, overwrite it
  86. fd, err := os.OpenFile(localpath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
  87. if err != nil {
  88. return resp, err
  89. }
  90. _, err = io.Copy(fd, resp.Body)
  91. fd.Close()
  92. if err != nil {
  93. return resp, err
  94. }
  95. return resp, nil
  96. }
  97. func (s *ObjectService) GetObjectURL(name string) *url.URL {
  98. uri, _ := url.Parse("/" + encodeURIComponent(name, []byte{'/'}))
  99. return s.client.BaseURL.BucketURL.ResolveReference(uri)
  100. }
  101. type PresignedURLOptions struct {
  102. Query *url.Values `xml:"-" url:"-" header:"-"`
  103. Header *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  104. }
  105. // GetPresignedURL get the object presigned to down or upload file by url
  106. func (s *ObjectService) GetPresignedURL(ctx context.Context, httpMethod, name, ak, sk string, expired time.Duration, opt interface{}) (*url.URL, error) {
  107. sendOpt := sendOptions{
  108. baseURL: s.client.BaseURL.BucketURL,
  109. uri: "/" + encodeURIComponent(name),
  110. method: httpMethod,
  111. optQuery: opt,
  112. optHeader: opt,
  113. }
  114. if popt, ok := opt.(*PresignedURLOptions); ok {
  115. qs := popt.Query.Encode()
  116. if qs != "" {
  117. sendOpt.uri = fmt.Sprintf("%s?%s", sendOpt.uri, qs)
  118. }
  119. }
  120. req, err := s.client.newRequest(ctx, sendOpt.baseURL, sendOpt.uri, sendOpt.method, sendOpt.body, sendOpt.optQuery, sendOpt.optHeader)
  121. if err != nil {
  122. return nil, err
  123. }
  124. var authTime *AuthTime
  125. if opt != nil {
  126. if opt, ok := opt.(*presignedURLTestingOptions); ok {
  127. authTime = opt.authTime
  128. }
  129. }
  130. if authTime == nil {
  131. authTime = NewAuthTime(expired)
  132. }
  133. authorization := newAuthorization(ak, sk, req, authTime)
  134. sign := encodeURIComponent(authorization, []byte{'&', '='})
  135. if req.URL.RawQuery == "" {
  136. req.URL.RawQuery = fmt.Sprintf("%s", sign)
  137. } else {
  138. req.URL.RawQuery = fmt.Sprintf("%s&%s", req.URL.RawQuery, sign)
  139. }
  140. return req.URL, nil
  141. }
  142. // ObjectPutHeaderOptions the options of header of the put object
  143. type ObjectPutHeaderOptions struct {
  144. CacheControl string `header:"Cache-Control,omitempty" url:"-"`
  145. ContentDisposition string `header:"Content-Disposition,omitempty" url:"-"`
  146. ContentEncoding string `header:"Content-Encoding,omitempty" url:"-"`
  147. ContentType string `header:"Content-Type,omitempty" url:"-"`
  148. ContentMD5 string `header:"Content-MD5,omitempty" url:"-"`
  149. ContentLength int64 `header:"Content-Length,omitempty" url:"-"`
  150. ContentLanguage string `header:"Content-Language,omitempty" url:"-"`
  151. Expect string `header:"Expect,omitempty" url:"-"`
  152. Expires string `header:"Expires,omitempty" url:"-"`
  153. XCosContentSHA1 string `header:"x-cos-content-sha1,omitempty" url:"-"`
  154. // 自定义的 x-cos-meta-* header
  155. XCosMetaXXX *http.Header `header:"x-cos-meta-*,omitempty" url:"-"`
  156. XCosStorageClass string `header:"x-cos-storage-class,omitempty" url:"-"`
  157. // 可选值: Normal, Appendable
  158. //XCosObjectType string `header:"x-cos-object-type,omitempty" url:"-"`
  159. // Enable Server Side Encryption, Only supported: AES256
  160. XCosServerSideEncryption string `header:"x-cos-server-side-encryption,omitempty" url:"-" xml:"-"`
  161. // SSE-C
  162. XCosSSECustomerAglo string `header:"x-cos-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  163. XCosSSECustomerKey string `header:"x-cos-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  164. XCosSSECustomerKeyMD5 string `header:"x-cos-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  165. //兼容其他自定义头部
  166. XOptionHeader *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  167. XCosTrafficLimit int `header:"x-cos-traffic-limit,omitempty" url:"-" xml:"-"`
  168. // 上传进度, ProgressCompleteEvent不能表示对应API调用成功,API是否调用成功的判断标准为返回err==nil
  169. Listener ProgressListener `header:"-" url:"-" xml:"-"`
  170. }
  171. // ObjectPutOptions the options of put object
  172. type ObjectPutOptions struct {
  173. *ACLHeaderOptions `header:",omitempty" url:"-" xml:"-"`
  174. *ObjectPutHeaderOptions `header:",omitempty" url:"-" xml:"-"`
  175. }
  176. // Put Object请求可以将一个文件(Oject)上传至指定Bucket。
  177. //
  178. // https://www.qcloud.com/document/product/436/7749
  179. func (s *ObjectService) Put(ctx context.Context, name string, r io.Reader, uopt *ObjectPutOptions) (*Response, error) {
  180. if r == nil {
  181. return nil, fmt.Errorf("reader is nil")
  182. }
  183. if err := CheckReaderLen(r); err != nil {
  184. return nil, err
  185. }
  186. opt := CloneObjectPutOptions(uopt)
  187. totalBytes, err := GetReaderLen(r)
  188. if err != nil && opt != nil && opt.Listener != nil {
  189. if opt.ContentLength == 0 {
  190. return nil, err
  191. }
  192. totalBytes = opt.ContentLength
  193. }
  194. if err == nil {
  195. // 与 go http 保持一致, 非bytes.Buffer/bytes.Reader/strings.Reader由用户指定ContentLength, 或使用 Chunk 上传
  196. if opt != nil && opt.ContentLength == 0 && IsLenReader(r) {
  197. opt.ContentLength = totalBytes
  198. }
  199. }
  200. reader := TeeReader(r, nil, totalBytes, nil)
  201. if s.client.Conf.EnableCRC {
  202. reader.writer = crc64.New(crc64.MakeTable(crc64.ECMA))
  203. }
  204. if opt != nil && opt.Listener != nil {
  205. reader.listener = opt.Listener
  206. }
  207. sendOpt := sendOptions{
  208. baseURL: s.client.BaseURL.BucketURL,
  209. uri: "/" + encodeURIComponent(name),
  210. method: http.MethodPut,
  211. body: reader,
  212. optHeader: opt,
  213. }
  214. resp, err := s.client.send(ctx, &sendOpt)
  215. return resp, err
  216. }
  217. // PutFromFile put object from local file
  218. func (s *ObjectService) PutFromFile(ctx context.Context, name string, filePath string, opt *ObjectPutOptions) (resp *Response, err error) {
  219. nr := 0
  220. for nr < 3 {
  221. fd, e := os.Open(filePath)
  222. if e != nil {
  223. err = e
  224. return
  225. }
  226. resp, err = s.Put(ctx, name, fd, opt)
  227. if err != nil {
  228. nr++
  229. fd.Close()
  230. continue
  231. }
  232. fd.Close()
  233. break
  234. }
  235. return
  236. }
  237. // ObjectCopyHeaderOptions is the head option of the Copy
  238. type ObjectCopyHeaderOptions struct {
  239. // When use replace directive to update meta infos
  240. CacheControl string `header:"Cache-Control,omitempty" url:"-"`
  241. ContentDisposition string `header:"Content-Disposition,omitempty" url:"-"`
  242. ContentEncoding string `header:"Content-Encoding,omitempty" url:"-"`
  243. ContentLanguage string `header:"Content-Language,omitempty" url:"-"`
  244. ContentType string `header:"Content-Type,omitempty" url:"-"`
  245. Expires string `header:"Expires,omitempty" url:"-"`
  246. Expect string `header:"Expect,omitempty" url:"-"`
  247. XCosMetadataDirective string `header:"x-cos-metadata-directive,omitempty" url:"-" xml:"-"`
  248. XCosCopySourceIfModifiedSince string `header:"x-cos-copy-source-If-Modified-Since,omitempty" url:"-" xml:"-"`
  249. XCosCopySourceIfUnmodifiedSince string `header:"x-cos-copy-source-If-Unmodified-Since,omitempty" url:"-" xml:"-"`
  250. XCosCopySourceIfMatch string `header:"x-cos-copy-source-If-Match,omitempty" url:"-" xml:"-"`
  251. XCosCopySourceIfNoneMatch string `header:"x-cos-copy-source-If-None-Match,omitempty" url:"-" xml:"-"`
  252. XCosStorageClass string `header:"x-cos-storage-class,omitempty" url:"-" xml:"-"`
  253. // 自定义的 x-cos-meta-* header
  254. XCosMetaXXX *http.Header `header:"x-cos-meta-*,omitempty" url:"-"`
  255. XCosCopySource string `header:"x-cos-copy-source" url:"-" xml:"-"`
  256. XCosServerSideEncryption string `header:"x-cos-server-side-encryption,omitempty" url:"-" xml:"-"`
  257. // SSE-C
  258. XCosSSECustomerAglo string `header:"x-cos-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  259. XCosSSECustomerKey string `header:"x-cos-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  260. XCosSSECustomerKeyMD5 string `header:"x-cos-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  261. XCosCopySourceSSECustomerAglo string `header:"x-cos-copy-source-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  262. XCosCopySourceSSECustomerKey string `header:"x-cos-copy-source-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  263. XCosCopySourceSSECustomerKeyMD5 string `header:"x-cos-copy-source-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  264. //兼容其他自定义头部
  265. XOptionHeader *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  266. }
  267. // ObjectCopyOptions is the option of Copy, choose header or body
  268. type ObjectCopyOptions struct {
  269. *ObjectCopyHeaderOptions `header:",omitempty" url:"-" xml:"-"`
  270. *ACLHeaderOptions `header:",omitempty" url:"-" xml:"-"`
  271. }
  272. // ObjectCopyResult is the result of Copy
  273. type ObjectCopyResult struct {
  274. XMLName xml.Name `xml:"CopyObjectResult"`
  275. ETag string `xml:"ETag,omitempty"`
  276. LastModified string `xml:"LastModified,omitempty"`
  277. CRC64 string `xml:"CRC64,omitempty"`
  278. VersionId string `xml:"VersionId,omitempty"`
  279. }
  280. // Copy 调用 PutObjectCopy 请求实现将一个文件从源路径复制到目标路径。建议文件大小 1M 到 5G,
  281. // 超过 5G 的文件请使用分块上传 Upload - Copy。在拷贝的过程中,文件元属性和 ACL 可以被修改。
  282. //
  283. // 用户可以通过该接口实现文件移动,文件重命名,修改文件属性和创建副本。
  284. //
  285. // 注意:在跨帐号复制的时候,需要先设置被复制文件的权限为公有读,或者对目标帐号赋权,同帐号则不需要。
  286. //
  287. // https://cloud.tencent.com/document/product/436/10881
  288. func (s *ObjectService) Copy(ctx context.Context, name, sourceURL string, opt *ObjectCopyOptions, id ...string) (*ObjectCopyResult, *Response, error) {
  289. surl := strings.SplitN(sourceURL, "/", 2)
  290. if len(surl) < 2 {
  291. return nil, nil, errors.New(fmt.Sprintf("x-cos-copy-source format error: %s", sourceURL))
  292. }
  293. var u string
  294. if len(id) == 1 {
  295. u = fmt.Sprintf("%s/%s?versionId=%s", surl[0], encodeURIComponent(surl[1]), id[0])
  296. } else if len(id) == 0 {
  297. u = fmt.Sprintf("%s/%s", surl[0], encodeURIComponent(surl[1]))
  298. } else {
  299. return nil, nil, errors.New("wrong params")
  300. }
  301. var res ObjectCopyResult
  302. copyOpt := &ObjectCopyOptions{
  303. &ObjectCopyHeaderOptions{},
  304. &ACLHeaderOptions{},
  305. }
  306. if opt != nil {
  307. if opt.ObjectCopyHeaderOptions != nil {
  308. *copyOpt.ObjectCopyHeaderOptions = *opt.ObjectCopyHeaderOptions
  309. }
  310. if opt.ACLHeaderOptions != nil {
  311. *copyOpt.ACLHeaderOptions = *opt.ACLHeaderOptions
  312. }
  313. }
  314. copyOpt.XCosCopySource = u
  315. sendOpt := sendOptions{
  316. baseURL: s.client.BaseURL.BucketURL,
  317. uri: "/" + encodeURIComponent(name),
  318. method: http.MethodPut,
  319. body: nil,
  320. optHeader: copyOpt,
  321. result: &res,
  322. }
  323. resp, err := s.client.doRetry(ctx, &sendOpt)
  324. // If the error occurs during the copy operation, the error response is embedded in the 200 OK response. This means that a 200 OK response can contain either a success or an error.
  325. if err == nil && resp.StatusCode == 200 {
  326. if res.ETag == "" {
  327. return &res, resp, errors.New("response 200 OK, but body contains an error")
  328. }
  329. }
  330. return &res, resp, err
  331. }
  332. type ObjectDeleteOptions struct {
  333. // SSE-C
  334. XCosSSECustomerAglo string `header:"x-cos-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  335. XCosSSECustomerKey string `header:"x-cos-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  336. XCosSSECustomerKeyMD5 string `header:"x-cos-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  337. //兼容其他自定义头部
  338. XOptionHeader *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  339. VersionId string `header:"-" url:"VersionId,omitempty" xml:"-"`
  340. }
  341. // Delete Object请求可以将一个文件(Object)删除。
  342. //
  343. // https://www.qcloud.com/document/product/436/7743
  344. func (s *ObjectService) Delete(ctx context.Context, name string, opt ...*ObjectDeleteOptions) (*Response, error) {
  345. var optHeader *ObjectDeleteOptions
  346. // When use "" string might call the delete bucket interface
  347. if len(name) == 0 {
  348. return nil, errors.New("empty object name")
  349. }
  350. if len(opt) > 0 {
  351. optHeader = opt[0]
  352. }
  353. sendOpt := sendOptions{
  354. baseURL: s.client.BaseURL.BucketURL,
  355. uri: "/" + encodeURIComponent(name),
  356. method: http.MethodDelete,
  357. optHeader: optHeader,
  358. optQuery: optHeader,
  359. }
  360. resp, err := s.client.doRetry(ctx, &sendOpt)
  361. return resp, err
  362. }
  363. // ObjectHeadOptions is the option of HeadObject
  364. type ObjectHeadOptions struct {
  365. IfModifiedSince string `url:"-" header:"If-Modified-Since,omitempty"`
  366. // SSE-C
  367. XCosSSECustomerAglo string `header:"x-cos-server-side-encryption-customer-algorithm,omitempty" url:"-" xml:"-"`
  368. XCosSSECustomerKey string `header:"x-cos-server-side-encryption-customer-key,omitempty" url:"-" xml:"-"`
  369. XCosSSECustomerKeyMD5 string `header:"x-cos-server-side-encryption-customer-key-MD5,omitempty" url:"-" xml:"-"`
  370. XOptionHeader *http.Header `header:"-,omitempty" url:"-" xml:"-"`
  371. }
  372. // Head Object请求可以取回对应Object的元数据,Head的权限与Get的权限一致
  373. //
  374. // https://www.qcloud.com/document/product/436/7745
  375. func (s *ObjectService) Head(ctx context.Context, name string, opt *ObjectHeadOptions, id ...string) (*Response, error) {
  376. var u string
  377. if len(id) == 1 {
  378. u = fmt.Sprintf("/%s?versionId=%s", encodeURIComponent(name), id[0])
  379. } else if len(id) == 0 {
  380. u = "/" + encodeURIComponent(name)
  381. } else {
  382. return nil, errors.New("wrong params")
  383. }
  384. sendOpt := sendOptions{
  385. baseURL: s.client.BaseURL.BucketURL,
  386. uri: u,
  387. method: http.MethodHead,
  388. optHeader: opt,
  389. }
  390. resp, err := s.client.doRetry(ctx, &sendOpt)
  391. if resp != nil && resp.Header["X-Cos-Object-Type"] != nil && resp.Header["X-Cos-Object-Type"][0] == "appendable" {
  392. resp.Header.Add("x-cos-next-append-position", resp.Header["Content-Length"][0])
  393. }
  394. return resp, err
  395. }
  396. // ObjectOptionsOptions is the option of object options
  397. type ObjectOptionsOptions struct {
  398. Origin string `url:"-" header:"Origin"`
  399. AccessControlRequestMethod string `url:"-" header:"Access-Control-Request-Method"`
  400. AccessControlRequestHeaders string `url:"-" header:"Access-Control-Request-Headers,omitempty"`
  401. }
  402. // Options Object请求实现跨域访问的预请求。即发出一个 OPTIONS 请求给服务器以确认是否可以进行跨域操作。
  403. //
  404. // 当CORS配置不存在时,请求返回403 Forbidden。
  405. //
  406. // https://www.qcloud.com/document/product/436/8288
  407. func (s *ObjectService) Options(ctx context.Context, name string, opt *ObjectOptionsOptions) (*Response, error) {
  408. sendOpt := sendOptions{
  409. baseURL: s.client.BaseURL.BucketURL,
  410. uri: "/" + encodeURIComponent(name),
  411. method: http.MethodOptions,
  412. optHeader: opt,
  413. }
  414. resp, err := s.client.send(ctx, &sendOpt)
  415. return resp, err
  416. }
  417. // CASJobParameters support three way: Standard(in 35 hours), Expedited(quick way, in 15 mins), Bulk(in 5-12 hours_
  418. type CASJobParameters struct {
  419. Tier string `xml:"Tier" header:"-" url:"-"`
  420. }
  421. // ObjectRestoreOptions is the option of object restore
  422. type ObjectRestoreOptions struct {
  423. XMLName xml.Name `xml:"RestoreRequest" header:"-" url:"-"`
  424. Days int `xml:"Days" header:"-" url:"-"`
  425. Tier *CASJobParameters `xml:"CASJobParameters" header:"-" url:"-"`
  426. XOptionHeader *http.Header `xml:"-" header:",omitempty" url:"-"`
  427. }
  428. // PutRestore API can recover an object of type archived by COS archive.
  429. //
  430. // https://cloud.tencent.com/document/product/436/12633
  431. func (s *ObjectService) PostRestore(ctx context.Context, name string, opt *ObjectRestoreOptions) (*Response, error) {
  432. u := fmt.Sprintf("/%s?restore", encodeURIComponent(name))
  433. sendOpt := sendOptions{
  434. baseURL: s.client.BaseURL.BucketURL,
  435. uri: u,
  436. method: http.MethodPost,
  437. body: opt,
  438. optHeader: opt,
  439. }
  440. resp, err := s.client.doRetry(ctx, &sendOpt)
  441. return resp, err
  442. }
  443. // Append请求可以将一个文件(Object)以分块追加的方式上传至 Bucket 中。使用Append Upload的文件必须事前被设定为Appendable。
  444. // 当Appendable的文件被执行Put Object的操作以后,文件被覆盖,属性改变为Normal。
  445. //
  446. // 文件属性可以在Head Object操作中被查询到,当您发起Head Object请求时,会返回自定义Header『x-cos-object-type』,该Header只有两个枚举值:Normal或者Appendable。
  447. //
  448. // 追加上传建议文件大小1M - 5G。如果position的值和当前Object的长度不致,COS会返回409错误。
  449. // 如果Append一个Normal的Object,COS会返回409 ObjectNotAppendable。
  450. //
  451. // Appendable的文件不可以被复制,不参与版本管理,不参与生命周期管理,不可跨区域复制。
  452. //
  453. // 当 r 不是 bytes.Buffer/bytes.Reader/strings.Reader 时,必须指定 opt.ObjectPutHeaderOptions.ContentLength
  454. //
  455. // https://www.qcloud.com/document/product/436/7741
  456. func (s *ObjectService) Append(ctx context.Context, name string, position int, r io.Reader, opt *ObjectPutOptions) (int, *Response, error) {
  457. res := position
  458. if r == nil {
  459. return res, nil, fmt.Errorf("reader is nil")
  460. }
  461. if err := CheckReaderLen(r); err != nil {
  462. return res, nil, err
  463. }
  464. opt = CloneObjectPutOptions(opt)
  465. totalBytes, err := GetReaderLen(r)
  466. if err != nil && opt != nil && opt.Listener != nil {
  467. if opt.ContentLength == 0 {
  468. return res, nil, err
  469. }
  470. totalBytes = opt.ContentLength
  471. }
  472. if err == nil {
  473. // 与 go http 保持一致, 非bytes.Buffer/bytes.Reader/strings.Reader需用户指定ContentLength
  474. if opt != nil && opt.ContentLength == 0 && IsLenReader(r) {
  475. opt.ContentLength = totalBytes
  476. }
  477. }
  478. reader := TeeReader(r, nil, totalBytes, nil)
  479. if s.client.Conf.EnableCRC {
  480. reader.writer = md5.New() // MD5校验
  481. reader.disableCheckSum = true
  482. }
  483. if opt != nil && opt.Listener != nil {
  484. reader.listener = opt.Listener
  485. }
  486. u := fmt.Sprintf("/%s?append&position=%d", encodeURIComponent(name), position)
  487. sendOpt := sendOptions{
  488. baseURL: s.client.BaseURL.BucketURL,
  489. uri: u,
  490. method: http.MethodPost,
  491. optHeader: opt,
  492. body: reader,
  493. }
  494. resp, err := s.client.send(ctx, &sendOpt)
  495. if err == nil {
  496. // 数据校验
  497. if s.client.Conf.EnableCRC && reader.writer != nil {
  498. wanted := hex.EncodeToString(reader.Sum())
  499. if wanted != resp.Header.Get("x-cos-content-sha1") {
  500. return res, resp, fmt.Errorf("append verification failed, want:%v, return:%v", wanted, resp.Header.Get("x-cos-content-sha1"))
  501. }
  502. }
  503. np, err := strconv.ParseInt(resp.Header.Get("x-cos-next-append-position"), 10, 64)
  504. return int(np), resp, err
  505. }
  506. return res, resp, err
  507. }
  508. // ObjectDeleteMultiOptions is the option of DeleteMulti
  509. type ObjectDeleteMultiOptions struct {
  510. XMLName xml.Name `xml:"Delete" header:"-"`
  511. Quiet bool `xml:"Quiet" header:"-"`
  512. Objects []Object `xml:"Object" header:"-"`
  513. //XCosSha1 string `xml:"-" header:"x-cos-sha1"`
  514. }
  515. // ObjectDeleteMultiResult is the result of DeleteMulti
  516. type ObjectDeleteMultiResult struct {
  517. XMLName xml.Name `xml:"DeleteResult"`
  518. DeletedObjects []Object `xml:"Deleted,omitempty"`
  519. Errors []struct {
  520. Key string `xml:",omitempty"`
  521. Code string `xml:",omitempty"`
  522. Message string `xml:",omitempty"`
  523. VersionId string `xml:",omitempty"`
  524. } `xml:"Error,omitempty"`
  525. }
  526. // DeleteMulti 请求实现批量删除文件,最大支持单次删除1000个文件。
  527. // 对于返回结果,COS提供Verbose和Quiet两种结果模式。Verbose模式将返回每个Object的删除结果;
  528. // Quiet模式只返回报错的Object信息。
  529. // https://www.qcloud.com/document/product/436/8289
  530. func (s *ObjectService) DeleteMulti(ctx context.Context, opt *ObjectDeleteMultiOptions) (*ObjectDeleteMultiResult, *Response, error) {
  531. var res ObjectDeleteMultiResult
  532. sendOpt := sendOptions{
  533. baseURL: s.client.BaseURL.BucketURL,
  534. uri: "/?delete",
  535. method: http.MethodPost,
  536. body: opt,
  537. result: &res,
  538. }
  539. resp, err := s.client.doRetry(ctx, &sendOpt)
  540. return &res, resp, err
  541. }
  542. // Object is the meta info of the object
  543. type Object struct {
  544. Key string `xml:",omitempty"`
  545. ETag string `xml:",omitempty"`
  546. Size int64 `xml:",omitempty"`
  547. PartNumber int `xml:",omitempty"`
  548. LastModified string `xml:",omitempty"`
  549. StorageClass string `xml:",omitempty"`
  550. Owner *Owner `xml:",omitempty"`
  551. VersionId string `xml:",omitempty"`
  552. }
  553. // MultiUploadOptions is the option of the multiupload,
  554. // ThreadPoolSize default is one
  555. type MultiUploadOptions struct {
  556. OptIni *InitiateMultipartUploadOptions
  557. PartSize int64
  558. ThreadPoolSize int
  559. CheckPoint bool
  560. EnableVerification bool
  561. }
  562. type MultiDownloadOptions struct {
  563. Opt *ObjectGetOptions
  564. PartSize int64
  565. ThreadPoolSize int
  566. CheckPoint bool
  567. CheckPointFile string
  568. }
  569. type MultiDownloadCPInfo struct {
  570. Size int64 `json:"contentLength,omitempty"`
  571. ETag string `json:"eTag,omitempty"`
  572. CRC64 string `json:"crc64ecma,omitempty"`
  573. LastModified string `json:"lastModified,omitempty"`
  574. DownloadedBlocks []DownloadedBlock `json:"downloadedBlocks,omitempty"`
  575. }
  576. type DownloadedBlock struct {
  577. From int64 `json:"from,omitempty"`
  578. To int64 `json:"to,omitempty"`
  579. }
  580. type Chunk struct {
  581. Number int
  582. OffSet int64
  583. Size int64
  584. Done bool
  585. ETag string
  586. }
  587. // jobs
  588. type Jobs struct {
  589. Name string
  590. UploadId string
  591. FilePath string
  592. RetryTimes int
  593. VersionId []string
  594. Chunk Chunk
  595. Data io.Reader
  596. Opt *ObjectUploadPartOptions
  597. DownOpt *ObjectGetOptions
  598. }
  599. type Results struct {
  600. PartNumber int
  601. Resp *Response
  602. err error
  603. }
  604. func LimitReadCloser(r io.Reader, n int64) io.Reader {
  605. var lc LimitedReadCloser
  606. lc.R = r
  607. lc.N = n
  608. return &lc
  609. }
  610. type LimitedReadCloser struct {
  611. io.LimitedReader
  612. }
  613. func (lc *LimitedReadCloser) Close() error {
  614. if r, ok := lc.R.(io.ReadCloser); ok {
  615. return r.Close()
  616. }
  617. return nil
  618. }
  619. type DiscardReadCloser struct {
  620. RC io.ReadCloser
  621. Discard int
  622. }
  623. func (drc *DiscardReadCloser) Read(data []byte) (int, error) {
  624. n, err := drc.RC.Read(data)
  625. if drc.Discard == 0 || n <= 0 {
  626. return n, err
  627. }
  628. if n <= drc.Discard {
  629. drc.Discard -= n
  630. return 0, err
  631. }
  632. realLen := n - drc.Discard
  633. copy(data[0:realLen], data[drc.Discard:n])
  634. drc.Discard = 0
  635. return realLen, err
  636. }
  637. func (drc *DiscardReadCloser) Close() error {
  638. if rc, ok := drc.RC.(io.ReadCloser); ok {
  639. return rc.Close()
  640. }
  641. return nil
  642. }
  643. func worker(ctx context.Context, s *ObjectService, jobs <-chan *Jobs, results chan<- *Results) {
  644. for j := range jobs {
  645. j.Opt.ContentLength = j.Chunk.Size
  646. rt := j.RetryTimes
  647. for {
  648. // http.Request.Body can be Closed in request
  649. fd, err := os.Open(j.FilePath)
  650. var res Results
  651. if err != nil {
  652. res.err = err
  653. res.PartNumber = j.Chunk.Number
  654. res.Resp = nil
  655. results <- &res
  656. break
  657. }
  658. fd.Seek(j.Chunk.OffSet, os.SEEK_SET)
  659. resp, err := s.UploadPart(ctx, j.Name, j.UploadId, j.Chunk.Number,
  660. LimitReadCloser(fd, j.Chunk.Size), j.Opt)
  661. res.PartNumber = j.Chunk.Number
  662. res.Resp = resp
  663. res.err = err
  664. if err != nil {
  665. rt--
  666. if rt == 0 {
  667. results <- &res
  668. break
  669. }
  670. continue
  671. }
  672. results <- &res
  673. break
  674. }
  675. }
  676. }
  677. func downloadWorker(ctx context.Context, s *ObjectService, jobs <-chan *Jobs, results chan<- *Results) {
  678. for j := range jobs {
  679. opt := &RangeOptions{
  680. HasStart: true,
  681. HasEnd: true,
  682. Start: j.Chunk.OffSet,
  683. End: j.Chunk.OffSet + j.Chunk.Size - 1,
  684. }
  685. j.DownOpt.Range = FormatRangeOptions(opt)
  686. rt := j.RetryTimes
  687. for {
  688. var res Results
  689. res.PartNumber = j.Chunk.Number
  690. resp, err := s.Get(ctx, j.Name, j.DownOpt, j.VersionId...)
  691. res.err = err
  692. res.Resp = resp
  693. if err != nil {
  694. rt--
  695. if rt == 0 {
  696. results <- &res
  697. break
  698. }
  699. continue
  700. }
  701. defer resp.Body.Close()
  702. fd, err := os.OpenFile(j.FilePath, os.O_WRONLY, 0660)
  703. if err != nil {
  704. res.err = err
  705. results <- &res
  706. break
  707. }
  708. fd.Seek(j.Chunk.OffSet, os.SEEK_SET)
  709. n, err := io.Copy(fd, LimitReadCloser(resp.Body, j.Chunk.Size))
  710. if n != j.Chunk.Size || err != nil {
  711. res.err = fmt.Errorf("io.Copy Failed, nread:%v, want:%v, err:%v", n, j.Chunk.Size, err)
  712. }
  713. fd.Close()
  714. results <- &res
  715. break
  716. }
  717. }
  718. }
  719. func DividePart(fileSize int64, last int) (int64, int64) {
  720. partSize := int64(last * 1024 * 1024)
  721. partNum := fileSize / partSize
  722. for partNum >= 10000 {
  723. partSize = partSize * 2
  724. partNum = fileSize / partSize
  725. }
  726. return partNum, partSize
  727. }
  728. func SplitFileIntoChunks(filePath string, partSize int64) (int64, []Chunk, int, error) {
  729. if filePath == "" {
  730. return 0, nil, 0, errors.New("filePath invalid")
  731. }
  732. file, err := os.Open(filePath)
  733. if err != nil {
  734. return 0, nil, 0, err
  735. }
  736. defer file.Close()
  737. stat, err := file.Stat()
  738. if err != nil {
  739. return 0, nil, 0, err
  740. }
  741. var partNum int64
  742. if partSize > 0 {
  743. if partSize < 1024*1024 {
  744. return 0, nil, 0, errors.New("partSize>=1048576 is required")
  745. }
  746. partNum = stat.Size() / partSize
  747. if partNum >= 10000 {
  748. return 0, nil, 0, errors.New("Too many parts, out of 10000")
  749. }
  750. } else {
  751. partNum, partSize = DividePart(stat.Size(), 64)
  752. }
  753. var chunks []Chunk
  754. var chunk = Chunk{}
  755. for i := int64(0); i < partNum; i++ {
  756. chunk.Number = int(i + 1)
  757. chunk.OffSet = i * partSize
  758. chunk.Size = partSize
  759. chunks = append(chunks, chunk)
  760. }
  761. if stat.Size()%partSize > 0 {
  762. chunk.Number = len(chunks) + 1
  763. chunk.OffSet = int64(len(chunks)) * partSize
  764. chunk.Size = stat.Size() % partSize
  765. chunks = append(chunks, chunk)
  766. partNum++
  767. }
  768. return int64(stat.Size()), chunks, int(partNum), nil
  769. }
  770. func (s *ObjectService) getResumableUploadID(ctx context.Context, name string) (string, error) {
  771. opt := &ObjectListUploadsOptions{
  772. Prefix: name,
  773. EncodingType: "url",
  774. }
  775. res, _, err := s.ListUploads(ctx, opt)
  776. if err != nil {
  777. return "", err
  778. }
  779. if len(res.Upload) == 0 {
  780. return "", nil
  781. }
  782. last := len(res.Upload) - 1
  783. for last >= 0 {
  784. decodeKey, _ := decodeURIComponent(res.Upload[last].Key)
  785. if decodeKey == name {
  786. return decodeURIComponent(res.Upload[last].UploadID)
  787. }
  788. last = last - 1
  789. }
  790. return "", nil
  791. }
  792. func (s *ObjectService) checkUploadedParts(ctx context.Context, name, UploadID, filepath string, chunks []Chunk, partNum int) error {
  793. var uploadedParts []Object
  794. isTruncated := true
  795. opt := &ObjectListPartsOptions{
  796. EncodingType: "url",
  797. }
  798. for isTruncated {
  799. res, _, err := s.ListParts(ctx, name, UploadID, opt)
  800. if err != nil {
  801. return err
  802. }
  803. if len(res.Parts) > 0 {
  804. uploadedParts = append(uploadedParts, res.Parts...)
  805. }
  806. isTruncated = res.IsTruncated
  807. opt.PartNumberMarker = res.NextPartNumberMarker
  808. }
  809. fd, err := os.Open(filepath)
  810. if err != nil {
  811. return err
  812. }
  813. defer fd.Close()
  814. // 某个分块出错, 重置chunks
  815. ret := func(e error) error {
  816. for i, _ := range chunks {
  817. chunks[i].Done = false
  818. chunks[i].ETag = ""
  819. }
  820. return e
  821. }
  822. for _, part := range uploadedParts {
  823. partNumber := part.PartNumber
  824. if partNumber > partNum {
  825. return ret(errors.New("Part Number is not consistent"))
  826. }
  827. partNumber = partNumber - 1
  828. fd.Seek(chunks[partNumber].OffSet, os.SEEK_SET)
  829. bs, err := ioutil.ReadAll(io.LimitReader(fd, chunks[partNumber].Size))
  830. if err != nil {
  831. return ret(err)
  832. }
  833. localMD5 := fmt.Sprintf("\"%x\"", md5.Sum(bs))
  834. if localMD5 != part.ETag {
  835. return ret(errors.New(fmt.Sprintf("CheckSum Failed in Part[%d]", part.PartNumber)))
  836. }
  837. chunks[partNumber].Done = true
  838. chunks[partNumber].ETag = part.ETag
  839. }
  840. return nil
  841. }
  842. // MultiUpload/Upload 为高级upload接口,并发分块上传
  843. //
  844. // 当 partSize > 0 时,由调用者指定分块大小,否则由 SDK 自动切分,单位为MB
  845. // 由调用者指定分块大小时,请确认分块数量不超过10000
  846. //
  847. func (s *ObjectService) MultiUpload(ctx context.Context, name string, filepath string, opt *MultiUploadOptions) (*CompleteMultipartUploadResult, *Response, error) {
  848. return s.Upload(ctx, name, filepath, opt)
  849. }
  850. func (s *ObjectService) Upload(ctx context.Context, name string, filepath string, opt *MultiUploadOptions) (*CompleteMultipartUploadResult, *Response, error) {
  851. if opt == nil {
  852. opt = &MultiUploadOptions{}
  853. }
  854. var localcrc uint64
  855. // 1.Get the file chunk
  856. totalBytes, chunks, partNum, err := SplitFileIntoChunks(filepath, opt.PartSize*1024*1024)
  857. if err != nil {
  858. return nil, nil, err
  859. }
  860. // 校验
  861. if s.client.Conf.EnableCRC {
  862. fd, err := os.Open(filepath)
  863. if err != nil {
  864. return nil, nil, err
  865. }
  866. defer fd.Close()
  867. localcrc, err = calCRC64(fd)
  868. if err != nil {
  869. return nil, nil, err
  870. }
  871. }
  872. // filesize=0 , use simple upload
  873. if partNum == 0 || partNum == 1 {
  874. var opt0 *ObjectPutOptions
  875. if opt.OptIni != nil {
  876. opt0 = &ObjectPutOptions{
  877. opt.OptIni.ACLHeaderOptions,
  878. opt.OptIni.ObjectPutHeaderOptions,
  879. }
  880. }
  881. rsp, err := s.PutFromFile(ctx, name, filepath, opt0)
  882. if err != nil {
  883. return nil, rsp, err
  884. }
  885. result := &CompleteMultipartUploadResult{
  886. Location: fmt.Sprintf("%s/%s", s.client.BaseURL.BucketURL, name),
  887. Key: name,
  888. ETag: rsp.Header.Get("ETag"),
  889. }
  890. if rsp != nil && s.client.Conf.EnableCRC {
  891. scoscrc := rsp.Header.Get("x-cos-hash-crc64ecma")
  892. icoscrc, _ := strconv.ParseUint(scoscrc, 10, 64)
  893. if icoscrc != localcrc {
  894. return result, rsp, fmt.Errorf("verification failed, want:%v, return:%v", localcrc, icoscrc)
  895. }
  896. }
  897. return result, rsp, nil
  898. }
  899. var uploadID string
  900. resumableFlag := false
  901. if opt.CheckPoint {
  902. var err error
  903. uploadID, err = s.getResumableUploadID(ctx, name)
  904. if err == nil && uploadID != "" {
  905. err = s.checkUploadedParts(ctx, name, uploadID, filepath, chunks, partNum)
  906. resumableFlag = (err == nil)
  907. }
  908. }
  909. // 2.Init
  910. optini := opt.OptIni
  911. if !resumableFlag {
  912. res, _, err := s.InitiateMultipartUpload(ctx, name, optini)
  913. if err != nil {
  914. return nil, nil, err
  915. }
  916. uploadID = res.UploadID
  917. }
  918. var poolSize int
  919. if opt.ThreadPoolSize > 0 {
  920. poolSize = opt.ThreadPoolSize
  921. } else {
  922. // Default is one
  923. poolSize = 1
  924. }
  925. chjobs := make(chan *Jobs, 100)
  926. chresults := make(chan *Results, 10000)
  927. optcom := &CompleteMultipartUploadOptions{}
  928. // 3.Start worker
  929. for w := 1; w <= poolSize; w++ {
  930. go worker(ctx, s, chjobs, chresults)
  931. }
  932. // progress started event
  933. var listener ProgressListener
  934. var consumedBytes int64
  935. if opt.OptIni != nil {
  936. if opt.OptIni.ObjectPutHeaderOptions != nil {
  937. listener = opt.OptIni.Listener
  938. }
  939. optcom.XOptionHeader, _ = deliverInitOptions(opt.OptIni)
  940. }
  941. event := newProgressEvent(ProgressStartedEvent, 0, 0, totalBytes)
  942. progressCallback(listener, event)
  943. // 4.Push jobs
  944. go func() {
  945. for _, chunk := range chunks {
  946. if chunk.Done {
  947. continue
  948. }
  949. partOpt := &ObjectUploadPartOptions{}
  950. if optini != nil && optini.ObjectPutHeaderOptions != nil {
  951. partOpt.XCosSSECustomerAglo = optini.XCosSSECustomerAglo
  952. partOpt.XCosSSECustomerKey = optini.XCosSSECustomerKey
  953. partOpt.XCosSSECustomerKeyMD5 = optini.XCosSSECustomerKeyMD5
  954. partOpt.XCosTrafficLimit = optini.XCosTrafficLimit
  955. }
  956. job := &Jobs{
  957. Name: name,
  958. RetryTimes: 3,
  959. FilePath: filepath,
  960. UploadId: uploadID,
  961. Chunk: chunk,
  962. Opt: partOpt,
  963. }
  964. chjobs <- job
  965. }
  966. close(chjobs)
  967. }()
  968. // 5.Recv the resp etag to complete
  969. err = nil
  970. for i := 0; i < partNum; i++ {
  971. if chunks[i].Done {
  972. optcom.Parts = append(optcom.Parts, Object{
  973. PartNumber: chunks[i].Number, ETag: chunks[i].ETag},
  974. )
  975. if err == nil {
  976. consumedBytes += chunks[i].Size
  977. event = newProgressEvent(ProgressDataEvent, chunks[i].Size, consumedBytes, totalBytes)
  978. progressCallback(listener, event)
  979. }
  980. continue
  981. }
  982. res := <-chresults
  983. // Notice one part fail can not get the etag according.
  984. if res.Resp == nil || res.err != nil {
  985. // Some part already fail, can not to get the header inside.
  986. err = fmt.Errorf("UploadID %s, part %d failed to get resp content. error: %s", uploadID, res.PartNumber, res.err.Error())
  987. continue
  988. }
  989. // Notice one part fail can not get the etag according.
  990. etag := res.Resp.Header.Get("ETag")
  991. optcom.Parts = append(optcom.Parts, Object{
  992. PartNumber: res.PartNumber, ETag: etag},
  993. )
  994. if err == nil {
  995. consumedBytes += chunks[res.PartNumber-1].Size
  996. event = newProgressEvent(ProgressDataEvent, chunks[res.PartNumber-1].Size, consumedBytes, totalBytes)
  997. progressCallback(listener, event)
  998. }
  999. }
  1000. close(chresults)
  1001. if err != nil {
  1002. event = newProgressEvent(ProgressFailedEvent, 0, consumedBytes, totalBytes, err)
  1003. progressCallback(listener, event)
  1004. return nil, nil, err
  1005. }
  1006. sort.Sort(ObjectList(optcom.Parts))
  1007. event = newProgressEvent(ProgressCompletedEvent, 0, consumedBytes, totalBytes)
  1008. progressCallback(listener, event)
  1009. v, resp, err := s.CompleteMultipartUpload(context.Background(), name, uploadID, optcom)
  1010. if err != nil {
  1011. return v, resp, err
  1012. }
  1013. if resp != nil && s.client.Conf.EnableCRC {
  1014. scoscrc := resp.Header.Get("x-cos-hash-crc64ecma")
  1015. icoscrc, _ := strconv.ParseUint(scoscrc, 10, 64)
  1016. if icoscrc != localcrc {
  1017. return v, resp, fmt.Errorf("verification failed, want:%v, return:%v", localcrc, icoscrc)
  1018. }
  1019. }
  1020. return v, resp, err
  1021. }
  1022. func SplitSizeIntoChunks(totalBytes int64, partSize int64) ([]Chunk, int, error) {
  1023. var partNum int64
  1024. if partSize > 0 {
  1025. if partSize < 1024*1024 {
  1026. return nil, 0, errors.New("partSize>=1048576 is required")
  1027. }
  1028. partNum = totalBytes / partSize
  1029. if partNum >= 10000 {
  1030. return nil, 0, errors.New("Too manry parts, out of 10000")
  1031. }
  1032. } else {
  1033. partNum, partSize = DividePart(totalBytes, 64)
  1034. }
  1035. var chunks []Chunk
  1036. var chunk = Chunk{}
  1037. for i := int64(0); i < partNum; i++ {
  1038. chunk.Number = int(i + 1)
  1039. chunk.OffSet = i * partSize
  1040. chunk.Size = partSize
  1041. chunks = append(chunks, chunk)
  1042. }
  1043. if totalBytes%partSize > 0 {
  1044. chunk.Number = len(chunks) + 1
  1045. chunk.OffSet = int64(len(chunks)) * partSize
  1046. chunk.Size = totalBytes % partSize
  1047. chunks = append(chunks, chunk)
  1048. partNum++
  1049. }
  1050. return chunks, int(partNum), nil
  1051. }
  1052. func (s *ObjectService) checkDownloadedParts(opt *MultiDownloadCPInfo, chfile string, chunks []Chunk) (*MultiDownloadCPInfo, bool) {
  1053. var defaultRes MultiDownloadCPInfo
  1054. defaultRes = *opt
  1055. fd, err := os.Open(chfile)
  1056. // checkpoint 文件不存在
  1057. if err != nil && os.IsNotExist(err) {
  1058. // 创建 checkpoint 文件
  1059. fd, _ = os.OpenFile(chfile, os.O_RDONLY|os.O_CREATE|os.O_TRUNC, 0660)
  1060. fd.Close()
  1061. return &defaultRes, false
  1062. }
  1063. if err != nil {
  1064. return &defaultRes, false
  1065. }
  1066. defer fd.Close()
  1067. var res MultiDownloadCPInfo
  1068. err = json.NewDecoder(fd).Decode(&res)
  1069. if err != nil {
  1070. return &defaultRes, false
  1071. }
  1072. // 与COS的文件比较
  1073. if res.CRC64 != opt.CRC64 || res.ETag != opt.ETag || res.Size != opt.Size || res.LastModified != opt.LastModified || len(res.DownloadedBlocks) == 0 {
  1074. return &defaultRes, false
  1075. }
  1076. // len(chunks) 大于1,否则为简单下载, chunks[0].Size为partSize
  1077. partSize := chunks[0].Size
  1078. for _, v := range res.DownloadedBlocks {
  1079. index := v.From / partSize
  1080. to := chunks[index].OffSet + chunks[index].Size - 1
  1081. if chunks[index].OffSet != v.From || to != v.To {
  1082. // 重置chunks
  1083. for i, _ := range chunks {
  1084. chunks[i].Done = false
  1085. }
  1086. return &defaultRes, false
  1087. }
  1088. chunks[index].Done = true
  1089. }
  1090. return &res, true
  1091. }
  1092. func (s *ObjectService) Download(ctx context.Context, name string, filepath string, opt *MultiDownloadOptions, id ...string) (*Response, error) {
  1093. // 参数校验
  1094. if opt == nil {
  1095. opt = &MultiDownloadOptions{}
  1096. }
  1097. if opt.Opt != nil && opt.Opt.Range != "" {
  1098. return nil, fmt.Errorf("Download doesn't support Range Options")
  1099. }
  1100. // 获取文件长度和CRC
  1101. var coscrc string
  1102. resp, err := s.Head(ctx, name, nil, id...)
  1103. if err != nil {
  1104. return resp, err
  1105. }
  1106. // 如果对象不存在x-cos-hash-crc64ecma,则跳过不做校验
  1107. coscrc = resp.Header.Get("x-cos-hash-crc64ecma")
  1108. strTotalBytes := resp.Header.Get("Content-Length")
  1109. totalBytes, err := strconv.ParseInt(strTotalBytes, 10, 64)
  1110. if err != nil {
  1111. return resp, err
  1112. }
  1113. // 切分
  1114. chunks, partNum, err := SplitSizeIntoChunks(totalBytes, opt.PartSize*1024*1024)
  1115. if err != nil {
  1116. return resp, err
  1117. }
  1118. // 直接下载到文件
  1119. if partNum == 0 || partNum == 1 {
  1120. rsp, err := s.GetToFile(ctx, name, filepath, opt.Opt, id...)
  1121. if err != nil {
  1122. return rsp, err
  1123. }
  1124. if coscrc != "" && s.client.Conf.EnableCRC {
  1125. icoscrc, _ := strconv.ParseUint(coscrc, 10, 64)
  1126. fd, err := os.Open(filepath)
  1127. if err != nil {
  1128. return rsp, err
  1129. }
  1130. defer fd.Close()
  1131. localcrc, err := calCRC64(fd)
  1132. if err != nil {
  1133. return rsp, err
  1134. }
  1135. if localcrc != icoscrc {
  1136. return rsp, fmt.Errorf("verification failed, want:%v, return:%v", icoscrc, localcrc)
  1137. }
  1138. }
  1139. return rsp, err
  1140. }
  1141. // 断点续载
  1142. var resumableFlag bool
  1143. var resumableInfo *MultiDownloadCPInfo
  1144. var cpfd *os.File
  1145. var cpfile string
  1146. if opt.CheckPoint {
  1147. cpInfo := &MultiDownloadCPInfo{
  1148. LastModified: resp.Header.Get("Last-Modified"),
  1149. ETag: resp.Header.Get("ETag"),
  1150. CRC64: coscrc,
  1151. Size: totalBytes,
  1152. }
  1153. cpfile = opt.CheckPointFile
  1154. if cpfile == "" {
  1155. cpfile = fmt.Sprintf("%s.cosresumabletask", filepath)
  1156. }
  1157. resumableInfo, resumableFlag = s.checkDownloadedParts(cpInfo, cpfile, chunks)
  1158. cpfd, err = os.OpenFile(cpfile, os.O_RDWR, 0660)
  1159. if err != nil {
  1160. return nil, fmt.Errorf("Open CheckPoint File[%v] Failed:%v", cpfile, err)
  1161. }
  1162. }
  1163. if !resumableFlag {
  1164. // 创建文件
  1165. nfile, err := os.OpenFile(filepath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
  1166. if err != nil {
  1167. if cpfd != nil {
  1168. cpfd.Close()
  1169. }
  1170. return resp, err
  1171. }
  1172. nfile.Close()
  1173. }
  1174. var poolSize int
  1175. if opt.ThreadPoolSize > 0 {
  1176. poolSize = opt.ThreadPoolSize
  1177. } else {
  1178. poolSize = 1
  1179. }
  1180. chjobs := make(chan *Jobs, 100)
  1181. chresults := make(chan *Results, 10000)
  1182. for w := 1; w <= poolSize; w++ {
  1183. go downloadWorker(ctx, s, chjobs, chresults)
  1184. }
  1185. go func() {
  1186. for _, chunk := range chunks {
  1187. if chunk.Done {
  1188. continue
  1189. }
  1190. var downOpt ObjectGetOptions
  1191. if opt.Opt != nil {
  1192. downOpt = *opt.Opt
  1193. downOpt.Listener = nil // listener need to set nil
  1194. }
  1195. job := &Jobs{
  1196. Name: name,
  1197. RetryTimes: 1,
  1198. FilePath: filepath,
  1199. Chunk: chunk,
  1200. DownOpt: &downOpt,
  1201. }
  1202. if len(id) > 0 {
  1203. job.VersionId = append(job.VersionId, id...)
  1204. }
  1205. chjobs <- job
  1206. }
  1207. close(chjobs)
  1208. }()
  1209. err = nil
  1210. for i := 0; i < partNum; i++ {
  1211. if chunks[i].Done {
  1212. continue
  1213. }
  1214. res := <-chresults
  1215. if res.Resp == nil || res.err != nil {
  1216. err = fmt.Errorf("part %d get resp Content. error: %s", res.PartNumber, res.err.Error())
  1217. continue
  1218. }
  1219. // Dump CheckPoint Info
  1220. if opt.CheckPoint {
  1221. cpfd.Truncate(0)
  1222. cpfd.Seek(0, os.SEEK_SET)
  1223. resumableInfo.DownloadedBlocks = append(resumableInfo.DownloadedBlocks, DownloadedBlock{
  1224. From: chunks[res.PartNumber-1].OffSet,
  1225. To: chunks[res.PartNumber-1].OffSet + chunks[res.PartNumber-1].Size - 1,
  1226. })
  1227. json.NewEncoder(cpfd).Encode(resumableInfo)
  1228. }
  1229. }
  1230. close(chresults)
  1231. if cpfd != nil {
  1232. cpfd.Close()
  1233. }
  1234. if err != nil {
  1235. return nil, err
  1236. }
  1237. // 下载成功,删除checkpoint文件
  1238. if opt.CheckPoint {
  1239. os.Remove(cpfile)
  1240. }
  1241. if coscrc != "" && s.client.Conf.EnableCRC {
  1242. icoscrc, _ := strconv.ParseUint(coscrc, 10, 64)
  1243. fd, err := os.Open(filepath)
  1244. if err != nil {
  1245. return resp, err
  1246. }
  1247. defer fd.Close()
  1248. localcrc, err := calCRC64(fd)
  1249. if err != nil {
  1250. return resp, err
  1251. }
  1252. if localcrc != icoscrc {
  1253. return resp, fmt.Errorf("verification failed, want:%v, return:%v", icoscrc, localcrc)
  1254. }
  1255. }
  1256. return resp, err
  1257. }
  1258. type ObjectPutTaggingOptions struct {
  1259. XMLName xml.Name `xml:"Tagging"`
  1260. TagSet []ObjectTaggingTag `xml:"TagSet>Tag,omitempty"`
  1261. }
  1262. type ObjectTaggingTag BucketTaggingTag
  1263. type ObjectGetTaggingResult ObjectPutTaggingOptions
  1264. func (s *ObjectService) PutTagging(ctx context.Context, name string, opt *ObjectPutTaggingOptions, id ...string) (*Response, error) {
  1265. var u string
  1266. if len(id) == 1 {
  1267. u = fmt.Sprintf("/%s?tagging&versionId=%s", encodeURIComponent(name), id[0])
  1268. } else if len(id) == 0 {
  1269. u = fmt.Sprintf("/%s?tagging", encodeURIComponent(name))
  1270. } else {
  1271. return nil, errors.New("wrong params")
  1272. }
  1273. sendOpt := &sendOptions{
  1274. baseURL: s.client.BaseURL.BucketURL,
  1275. uri: u,
  1276. method: http.MethodPut,
  1277. body: opt,
  1278. }
  1279. resp, err := s.client.doRetry(ctx, sendOpt)
  1280. return resp, err
  1281. }
  1282. func (s *ObjectService) GetTagging(ctx context.Context, name string, id ...string) (*ObjectGetTaggingResult, *Response, error) {
  1283. var u string
  1284. if len(id) == 1 {
  1285. u = fmt.Sprintf("/%s?tagging&versionId=%s", encodeURIComponent(name), id[0])
  1286. } else if len(id) == 0 {
  1287. u = fmt.Sprintf("/%s?tagging", encodeURIComponent(name))
  1288. } else {
  1289. return nil, nil, errors.New("wrong params")
  1290. }
  1291. var res ObjectGetTaggingResult
  1292. sendOpt := &sendOptions{
  1293. baseURL: s.client.BaseURL.BucketURL,
  1294. uri: u,
  1295. method: http.MethodGet,
  1296. result: &res,
  1297. }
  1298. resp, err := s.client.doRetry(ctx, sendOpt)
  1299. return &res, resp, err
  1300. }
  1301. func (s *ObjectService) DeleteTagging(ctx context.Context, name string, id ...string) (*Response, error) {
  1302. var u string
  1303. if len(id) == 1 {
  1304. u = fmt.Sprintf("/%s?tagging&versionId=%s", encodeURIComponent(name), id[0])
  1305. } else if len(id) == 0 {
  1306. u = fmt.Sprintf("/%s?tagging", encodeURIComponent(name))
  1307. } else {
  1308. return nil, errors.New("wrong params")
  1309. }
  1310. sendOpt := &sendOptions{
  1311. baseURL: s.client.BaseURL.BucketURL,
  1312. uri: u,
  1313. method: http.MethodDelete,
  1314. }
  1315. resp, err := s.client.doRetry(ctx, sendOpt)
  1316. return resp, err
  1317. }