func (d *downloader) setTotalBytes(resp *s3.GetObjectOutput) {
	d.m.Lock()
	defer d.m.Unlock()
	if d.totalBytes >= 0 {
		return
	}
	if resp.ContentRange == nil {
		if resp.ContentLength != nil {
			d.totalBytes = *resp.ContentLength
			return
		}
	} else {
		parts := strings.Split(*resp.ContentRange, "/")
		total := int64(-1)
		var err error
		totalStr := parts[len(parts)-1]
		if totalStr != "*" {
			total, err = strconv.ParseInt(totalStr, 10, 64)
			if err != nil {
				d.err = err
				return
			}
		}
		d.totalBytes = total
	}
}
setTotalBytes is a thread-safe setter for setting the total byte status.
Will extract the object's total bytes from the Content-Range if the file
will be chunked, or Content-Length. Content-Length is used when the response
does not include a Content-Range. Meaning the object was not chunked. This
occurs when the full file fits within the PartSize directive.
package s3manager
import (
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/awsutil"
	"github.com/aws/aws-sdk-go/aws/client"
	"github.com/aws/aws-sdk-go/aws/request"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
)
const DefaultDownloadPartSize = 1024 * 1024 * 5
const DefaultDownloadConcurrency = 5
type errReadingBody struct {
	err error
}
func (e *errReadingBody) Error() string {
	return fmt.Sprintf("failed to read part body: %v", e.err)
}
func (e *errReadingBody) Unwrap() error {
	return e.err
}
type Downloader struct {
	PartSize int64
	Concurrency int
	S3 s3iface.S3API
	RequestOptions []request.Option
	BufferProvider WriterReadFromProvider
}
func WithDownloaderRequestOptions(opts ...request.Option) func(*Downloader) {
	return func(d *Downloader) {
		d.RequestOptions = append(d.RequestOptions, opts...)
	}
}
func NewDownloader(c client.ConfigProvider, options ...func(*Downloader)) *Downloader {
	return newDownloader(s3.New(c), options...)
}
func newDownloader(client s3iface.S3API, options ...func(*Downloader)) *Downloader {
	d := &Downloader{
		S3:             client,
		PartSize:       DefaultDownloadPartSize,
		Concurrency:    DefaultDownloadConcurrency,
		BufferProvider: defaultDownloadBufferProvider(),
	}
	for _, option := range options {
		option(d)
	}
	return d
}
func NewDownloaderWithClient(svc s3iface.S3API, options ...func(*Downloader)) *Downloader {
	return newDownloader(svc, options...)
}
type maxRetrier interface {
	MaxRetries() int
}
func (d Downloader) Download(w io.WriterAt, input *s3.GetObjectInput, options ...func(*Downloader)) (n int64, err error) {
	return d.DownloadWithContext(aws.BackgroundContext(), w, input, options...)
}
func (d Downloader) DownloadWithContext(ctx aws.Context, w io.WriterAt, input *s3.GetObjectInput, options ...func(*Downloader)) (n int64, err error) {
	impl := downloader{w: w, in: input, cfg: d, ctx: ctx}
	for _, option := range options {
		option(&impl.cfg)
	}
	impl.cfg.RequestOptions = append(impl.cfg.RequestOptions, request.WithAppendUserAgent("S3Manager"))
	if s, ok := d.S3.(maxRetrier); ok {
		impl.partBodyMaxRetries = s.MaxRetries()
	}
	impl.totalBytes = -1
	if impl.cfg.Concurrency == 0 {
		impl.cfg.Concurrency = DefaultDownloadConcurrency
	}
	if impl.cfg.PartSize == 0 {
		impl.cfg.PartSize = DefaultDownloadPartSize
	}
	return impl.download()
}
func (d Downloader) DownloadWithIterator(ctx aws.Context, iter BatchDownloadIterator, opts ...func(*Downloader)) error {
	var errs []Error
	for iter.Next() {
		object := iter.DownloadObject()
		if _, err := d.DownloadWithContext(ctx, object.Writer, object.Object, opts...); err != nil {
			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
		}
		if object.After == nil {
			continue
		}
		if err := object.After(); err != nil {
			errs = append(errs, newError(err, object.Object.Bucket, object.Object.Key))
		}
	}
	if len(errs) > 0 {
		return NewBatchError("BatchedDownloadIncomplete", "some objects have failed to download.", errs)
	}
	return nil
}
type downloader struct {
	ctx aws.Context
	cfg Downloader
	in *s3.GetObjectInput
	w  io.WriterAt
	wg sync.WaitGroup
	m  sync.Mutex
	pos        int64
	totalBytes int64
	written    int64
	err        error
	partBodyMaxRetries int
}
func (d *downloader) download() (n int64, err error) {
	if rng := aws.StringValue(d.in.Range); len(rng) > 0 {
		d.downloadRange(rng)
		return d.written, d.err
	}
	d.getChunk()
	if total := d.getTotalBytes(); total >= 0 {
		ch := make(chan dlchunk, d.cfg.Concurrency)
		for i := 0; i < d.cfg.Concurrency; i++ {
			d.wg.Add(1)
			go d.downloadPart(ch)
		}
		for d.getErr() == nil {
			if d.pos >= total {
				break  
			}
			ch <- dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize}
			d.pos += d.cfg.PartSize
		}
		close(ch)
		d.wg.Wait()
	} else {
		for d.err == nil {
			d.getChunk()
		}
		e, ok := d.err.(awserr.RequestFailure)
		if ok && e.StatusCode() == http.StatusRequestedRangeNotSatisfiable {
			d.err = nil
		}
	}
	return d.written, d.err
}
func (d *downloader) downloadPart(ch chan dlchunk) {
	defer d.wg.Done()
	for {
		chunk, ok := <-ch
		if !ok {
			break
		}
		if d.getErr() != nil {
			continue
		}
		if err := d.downloadChunk(chunk); err != nil {
			d.setErr(err)
		}
	}
}
func (d *downloader) getChunk() {
	if d.getErr() != nil {
		return
	}
	chunk := dlchunk{w: d.w, start: d.pos, size: d.cfg.PartSize}
	d.pos += d.cfg.PartSize
	if err := d.downloadChunk(chunk); err != nil {
		d.setErr(err)
	}
}
func (d *downloader) downloadRange(rng string) {
	if d.getErr() != nil {
		return
	}
	chunk := dlchunk{w: d.w, start: d.pos}
	chunk.withRange = rng
	if err := d.downloadChunk(chunk); err != nil {
		d.setErr(err)
	}
	d.pos = d.written
}
func (d *downloader) downloadChunk(chunk dlchunk) error {
	in := &s3.GetObjectInput{}
	awsutil.Copy(in, d.in)
	in.Range = aws.String(chunk.ByteRange())
	var n int64
	var err error
	for retry := 0; retry <= d.partBodyMaxRetries; retry++ {
		n, err = d.tryDownloadChunk(in, &chunk)
		if err == nil {
			break
		}
		if bodyErr, ok := err.(*errReadingBody); ok {
			err = bodyErr.Unwrap()
		} else {
			return err
		}
		chunk.cur = 0
		logMessage(d.cfg.S3, aws.LogDebugWithRequestRetries,
			fmt.Sprintf("DEBUG: object part body download interrupted %s, err, %v, retrying attempt %d",
				aws.StringValue(in.Key), err, retry))
	}
	d.incrWritten(n)
	return err
}
func (d *downloader) tryDownloadChunk(in *s3.GetObjectInput, w io.Writer) (int64, error) {
	cleanup := func() {}
	if d.cfg.BufferProvider != nil {
		w, cleanup = d.cfg.BufferProvider.GetReadFrom(w)
	}
	defer cleanup()
	resp, err := d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...)
	if err != nil {
		return 0, err
	}
	d.setTotalBytes(resp)  
	n, err := io.Copy(w, resp.Body)
	resp.Body.Close()
	if err != nil {
		return n, &errReadingBody{err: err}
	}
	return n, nil
}
func logMessage(svc s3iface.S3API, level aws.LogLevelType, msg string) {
	s, ok := svc.(*s3.S3)
	if !ok {
		return
	}
	if s.Config.Logger == nil {
		return
	}
	if s.Config.LogLevel.Matches(level) {
		s.Config.Logger.Log(msg)
	}
}
func (d *downloader) getTotalBytes() int64 {
	d.m.Lock()
	defer d.m.Unlock()
	return d.totalBytes
}