func (m MultipartUpload) uploadPart(ch chan Event, chunk []byte, partNum int) (*s3.CompletedPart, error) {
	partInput := &s3.UploadPartInput{
		Body:          bytes.NewReader(chunk),
		Bucket:        m.res.Bucket,
		Key:           m.res.Key,
		PartNumber:    aws.Int64(int64(partNum)),
		UploadId:      m.res.UploadId,
		ContentLength: aws.Int64(int64(len(chunk))),
	}
	tryNum := 1
	for tryNum <= m.MaxRetries {
		res, err := m.svc.UploadPart(partInput)
		if err != nil {
			if tryNum == m.MaxRetries {
				if aerr, ok := err.(awserr.Error); ok {
					return nil, aerr
				}
				return nil, err
			}
			ch <- Retry{
				PartNumber:  m.currentPartNumber,
				RetryNumber: tryNum,
				MaxRetries:  m.MaxRetries,
			}
			tryNum++
		} else {
			return &s3.CompletedPart{
				ETag:       res.ETag,
				PartNumber: aws.Int64(int64(partNum)),
			}, nil
		}
	}
	return nil, errors.New("could not upload part")
}
uploadPart performs the technical S3 stuff to upload one part of the
multipart upload. If it fails we'll retry based on the number set in
multipartUploadManager.MaxRetries.
package pipedream
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
)
const (
	Kilobyte int64 = 1024
	Megabyte int64 = Kilobyte * 1024
	DefaultRegion = "us-east-1"
)
type Event interface {
	event()
}
type Progress struct {
	PartNumber int
	Bytes      int
}
type Retry struct {
	PartNumber  int
	RetryNumber int
	MaxRetries  int
}
type Complete struct {
	Bytes  int
	Result *s3.CompleteMultipartUploadOutput
}
type Error struct {
	Err error
}
func (e Error) Error() string {
	return e.Err.Error()
}
func (p Progress) event() {}
func (r Retry) event()    {}
func (c Complete) event() {}
func (e Error) event()    {}
type MultipartUpload struct {
	Endpoint    string
	Region      string
	Bucket      string
	AccessKey   string
	SecretKey   string
	MaxRetries  int
	MaxPartSize int64
	svc               *s3.S3
	res               *s3.CreateMultipartUploadOutput
	completedParts    []*s3.CompletedPart
	currentPartNumber int
	path              string
	reader            io.Reader
}
func (m *MultipartUpload) Send(reader io.Reader, path string) chan Event {
	m.reader = reader
	m.path = path
	ch := make(chan Event)
	go m.run(ch)
	return ch
}
func (m *MultipartUpload) run(ch chan Event) {
	if m.MaxRetries == 0 {
		m.MaxRetries = 3
	}
	if m.MaxPartSize == 0 {
		m.MaxPartSize = Megabyte * 5
	}
	if m.Endpoint == "" {
		m.Endpoint = "nyc3.digitaloceanspaces.com"
	}
	if m.Region == "" {
		m.Region = DefaultRegion
	}
	var missing []string
	if m.AccessKey == "" {
		missing = append(missing, "AccessKey")
	}
	if m.SecretKey == "" {
		missing = append(missing, "SecretKey")
	}
	if m.Bucket == "" {
		missing = append(missing, "Bucket")
	}
	if len(missing) > 0 {
		ch <- Error{
			Err: errors.New("missing " + EnglishJoin(missing, true)),
		}
		return
	}
	s3Config := &aws.Config{
		Credentials: credentials.NewStaticCredentials(m.AccessKey, m.SecretKey, ""),
		Endpoint:    aws.String(m.Endpoint),
		Region:      aws.String(m.Region),
	}
	newSession := session.New(s3Config)
	m.svc = s3.New(newSession)
	totalBytes := 0
	m.currentPartNumber = 1
	buf := make([]byte, m.MaxPartSize)
	for {
		n, err := m.reader.Read(buf)
		if err != nil && err == io.EOF {
			break
		}
		if err != nil {
			if abortErr := m.Abort(); abortErr != nil {
				ch <- Error{
					Err: fmt.Errorf("upload error: %v, as well as an error aborting the upload: %v", err, abortErr),
				}
				return
			}
			ch <- Error{err}
			return
		}
		if m.res == nil {
			input := &s3.CreateMultipartUploadInput{
				Bucket:      aws.String(m.Bucket),
				Key:         aws.String(m.path),
				ContentType: aws.String(http.DetectContentType(buf[:n])),
			}
			m.res, err = m.svc.CreateMultipartUpload(input)
			if err != nil {
				ch <- Error{err}
				return
			}
		}
		part, err := m.uploadPart(ch, buf[:n], m.currentPartNumber)
		if err != nil {
			if abortErr := m.Abort(); abortErr != nil {
				ch <- Error{
					Err: fmt.Errorf("upload error: %v, as well as an error aborting the upload: %v", err, abortErr),
				}
				return
			}
			ch <- Error{err}
			return
		}
		ch <- Progress{
			PartNumber: m.currentPartNumber,
			Bytes:      n,
		}
		totalBytes += n
		m.completedParts = append(m.completedParts, part)
		m.currentPartNumber++
	}
	res, err := m.complete()
	if err != nil {
		ch <- Error{err}
	}
	ch <- Complete{
		Bytes:  totalBytes,
		Result: res,
	}
}