| // Copyright 2016 The Go Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file. |
| |
| package gensupport |
| |
| import ( |
| "errors" |
| "fmt" |
| "io" |
| "net/http" |
| "sync" |
| "time" |
| |
| "golang.org/x/net/context" |
| ) |
| |
| const ( |
| // statusTooManyRequests is returned by the storage API if the |
| // per-project limits have been temporarily exceeded. The request |
| // should be retried. |
| // https://cloud.google.com/storage/docs/json_api/v1/status-codes#standardcodes |
| statusTooManyRequests = 429 |
| ) |
| |
| // ResumableUpload is used by the generated APIs to provide resumable uploads. |
| // It is not used by developers directly. |
| type ResumableUpload struct { |
| Client *http.Client |
| // URI is the resumable resource destination provided by the server after specifying "&uploadType=resumable". |
| URI string |
| UserAgent string // User-Agent for header of the request |
| // Media is the object being uploaded. |
| Media *MediaBuffer |
| // MediaType defines the media type, e.g. "image/jpeg". |
| MediaType string |
| |
| mu sync.Mutex // guards progress |
| progress int64 // number of bytes uploaded so far |
| |
| // Callback is an optional function that will be periodically called with the cumulative number of bytes uploaded. |
| Callback func(int64) |
| |
| // If not specified, a default exponential backoff strategy will be used. |
| Backoff BackoffStrategy |
| } |
| |
| // Progress returns the number of bytes uploaded at this point. |
| func (rx *ResumableUpload) Progress() int64 { |
| rx.mu.Lock() |
| defer rx.mu.Unlock() |
| return rx.progress |
| } |
| |
| // doUploadRequest performs a single HTTP request to upload data. |
| // off specifies the offset in rx.Media from which data is drawn. |
| // size is the number of bytes in data. |
| // final specifies whether data is the final chunk to be uploaded. |
| func (rx *ResumableUpload) doUploadRequest(ctx context.Context, data io.Reader, off, size int64, final bool) (*http.Response, error) { |
| req, err := http.NewRequest("POST", rx.URI, data) |
| if err != nil { |
| return nil, err |
| } |
| |
| req.ContentLength = size |
| var contentRange string |
| if final { |
| if size == 0 { |
| contentRange = fmt.Sprintf("bytes */%v", off) |
| } else { |
| contentRange = fmt.Sprintf("bytes %v-%v/%v", off, off+size-1, off+size) |
| } |
| } else { |
| contentRange = fmt.Sprintf("bytes %v-%v/*", off, off+size-1) |
| } |
| req.Header.Set("Content-Range", contentRange) |
| req.Header.Set("Content-Type", rx.MediaType) |
| req.Header.Set("User-Agent", rx.UserAgent) |
| |
| // Google's upload endpoint uses status code 308 for a |
| // different purpose than the "308 Permanent Redirect" |
| // since-standardized in RFC 7238. Because of the conflict in |
| // semantics, Google added this new request header which |
| // causes it to not use "308" and instead reply with 200 OK |
| // and sets the upload-specific "X-HTTP-Status-Code-Override: |
| // 308" response header. |
| req.Header.Set("X-GUploader-No-308", "yes") |
| |
| return SendRequest(ctx, rx.Client, req) |
| } |
| |
| func statusResumeIncomplete(resp *http.Response) bool { |
| // This is how the server signals "status resume incomplete" |
| // when X-GUploader-No-308 is set to "yes": |
| return resp != nil && resp.Header.Get("X-Http-Status-Code-Override") == "308" |
| } |
| |
| // reportProgress calls a user-supplied callback to report upload progress. |
| // If old==updated, the callback is not called. |
| func (rx *ResumableUpload) reportProgress(old, updated int64) { |
| if updated-old == 0 { |
| return |
| } |
| rx.mu.Lock() |
| rx.progress = updated |
| rx.mu.Unlock() |
| if rx.Callback != nil { |
| rx.Callback(updated) |
| } |
| } |
| |
| // transferChunk performs a single HTTP request to upload a single chunk from rx.Media. |
| func (rx *ResumableUpload) transferChunk(ctx context.Context) (*http.Response, error) { |
| chunk, off, size, err := rx.Media.Chunk() |
| |
| done := err == io.EOF |
| if !done && err != nil { |
| return nil, err |
| } |
| |
| res, err := rx.doUploadRequest(ctx, chunk, off, int64(size), done) |
| if err != nil { |
| return res, err |
| } |
| |
| // We sent "X-GUploader-No-308: yes" (see comment elsewhere in |
| // this file), so we don't expect to get a 308. |
| if res.StatusCode == 308 { |
| return nil, errors.New("unexpected 308 response status code") |
| } |
| |
| if res.StatusCode == http.StatusOK { |
| rx.reportProgress(off, off+int64(size)) |
| } |
| |
| if statusResumeIncomplete(res) { |
| rx.Media.Next() |
| } |
| return res, nil |
| } |
| |
| func contextDone(ctx context.Context) bool { |
| select { |
| case <-ctx.Done(): |
| return true |
| default: |
| return false |
| } |
| } |
| |
| // Upload starts the process of a resumable upload with a cancellable context. |
| // It retries using the provided back off strategy until cancelled or the |
| // strategy indicates to stop retrying. |
| // It is called from the auto-generated API code and is not visible to the user. |
| // Before sending an HTTP request, Upload calls any registered hook functions, |
| // and calls the returned functions after the request returns (see send.go). |
| // rx is private to the auto-generated API code. |
| // Exactly one of resp or err will be nil. If resp is non-nil, the caller must call resp.Body.Close. |
| func (rx *ResumableUpload) Upload(ctx context.Context) (resp *http.Response, err error) { |
| var pause time.Duration |
| backoff := rx.Backoff |
| if backoff == nil { |
| backoff = DefaultBackoffStrategy() |
| } |
| |
| for { |
| // Ensure that we return in the case of cancelled context, even if pause is 0. |
| if contextDone(ctx) { |
| return nil, ctx.Err() |
| } |
| select { |
| case <-ctx.Done(): |
| return nil, ctx.Err() |
| case <-time.After(pause): |
| } |
| |
| resp, err = rx.transferChunk(ctx) |
| |
| var status int |
| if resp != nil { |
| status = resp.StatusCode |
| } |
| |
| // Check if we should retry the request. |
| if shouldRetry(status, err) { |
| var retry bool |
| pause, retry = backoff.Pause() |
| if retry { |
| if resp != nil && resp.Body != nil { |
| resp.Body.Close() |
| } |
| continue |
| } |
| } |
| |
| // If the chunk was uploaded successfully, but there's still |
| // more to go, upload the next chunk without any delay. |
| if statusResumeIncomplete(resp) { |
| pause = 0 |
| backoff.Reset() |
| resp.Body.Close() |
| continue |
| } |
| |
| // It's possible for err and resp to both be non-nil here, but we expose a simpler |
| // contract to our callers: exactly one of resp and err will be non-nil. This means |
| // that any response body must be closed here before returning a non-nil error. |
| if err != nil { |
| if resp != nil && resp.Body != nil { |
| resp.Body.Close() |
| } |
| return nil, err |
| } |
| |
| return resp, nil |
| } |
| } |