Source: resumable.go in package google.golang.org/api/internal/gensupport


package gensupport

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"sync"
	"time"

	gax "github.com/googleapis/gax-go/v2"
)

Backoff is an interface around gax.Backoff's Pause method, allowing tests to provide their own implementation.

type Backoff interface {
	Pause() time.Duration
}

These are declared as global variables so that tests can overwrite them.

var (
	retryDeadline = 32 * time.Second
	backoff       = func() Backoff {
		return &gax.Backoff{Initial: 100 * time.Millisecond}

isRetryable is a platform-specific hook, specified in retryable_linux.go

	syscallRetryable func(error) bool = func(err error) bool { return false }
)

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 {

URI is the resumable resource destination provided by the server after specifying "&uploadType=resumable".

	URI       string

Media is the object being uploaded.

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)
}

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)
}

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
}

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) {

There are a couple of cases where it's possible for err and resp to both be non-nil. However, 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.

	var prepareReturn = func(resp *http.Response, err error) (*http.Response, error) {
		if err != nil {
			if resp != nil && resp.Body != nil {
				resp.Body.Close()
			}
			return nil, err
		}
		return resp, nil
	}

Send all chunks.

	for {
		var pause time.Duration

Each chunk gets its own initialized-at-zero retry.

		bo := backoff()
		quitAfter := time.After(retryDeadline)

Retry loop for a single chunk.

		for {
			select {
			case <-ctx.Done():
				if err == nil {
					err = ctx.Err()
				}
				return prepareReturn(resp, err)
			case <-time.After(pause):
			case <-quitAfter:
				return prepareReturn(resp, err)
			}

			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) {
				break
			}

			pause = bo.Pause()
			if resp != nil && resp.Body != nil {
				resp.Body.Close()
			}
		}

If the chunk was uploaded successfully, but there's still more to go, upload the next chunk without any delay.

		if statusResumeIncomplete(resp) {
			resp.Body.Close()
			continue
		}

		return prepareReturn(resp, err)
	}
}

shouldRetry indicates whether an error is retryable for the purposes of this package, following guidance from https://cloud.google.com/storage/docs/exponential-backoff .

func shouldRetry(status int, err error) bool {
	if 500 <= status && status <= 599 {
		return true
	}
	if status == statusTooManyRequests {
		return true
	}
	if err == io.ErrUnexpectedEOF {
		return true

Transient network errors should be retried.

	if syscallRetryable(err) {
		return true
	}
	if err, ok := err.(interface{ Temporary() bool }); ok {
		if err.Temporary() {
			return true
		}

If Go 1.13 error unwrapping is available, use this to examine wrapped errors.

	if err, ok := err.(interface{ Unwrap() error }); ok {
		return shouldRetry(status, err.Unwrap())
	}
	return false