s3upload.go

package main

import (
	"bytes"
	"compress/gzip"
	"fmt"
	"io"
	"os"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
)

type ZipPipe struct {
	Reader io.Reader
	pw     *io.PipeWriter
	Writer *gzip.Writer
}

func NewZipPipe() *ZipPipe {

	r, w := io.Pipe()
	zp := ZipPipe{
		Reader: r,
		pw:     w,
		Writer: gzip.NewWriter(w),
	}

	return &zp
}

func (zp *ZipPipe) Close() {
	zp.Writer.Flush()
	zp.Writer.Close()
	zp.pw.Close()
}

func (zp *ZipPipe) Flush() {
	zp.Writer.Flush()
}

//S3Upload contains everything needed to stream an object
type S3Upload struct {
	key    string
	bucket string

	uploader *s3manager.Uploader
	Buff     bytes.Buffer
	ZP       *ZipPipe
}

// NewS3Upload returns a S3Upload
func NewS3Upload(bucket string, key string) *S3Upload {

	s3u := S3Upload{}

	s3u.bucket = bucket
	s3u.key = key

	//s3u.r, s3u.Writer = io.Pipe()
	//s3u.Buff = bytes.NewBuffer(make([]byte, 0, 1024*1024*20))
	s3u.ZP = NewZipPipe()

	//s3u.Writer = gzip.NewWriter(s3u.Buff)
	// The session the S3 Uploader will use
	sess := session.Must(session.NewSession())

	// Create an uploader with the session and default options
	s3u.uploader = s3manager.NewUploader(sess, func(u *s3manager.Uploader) {
		// Define a strategy that will buffer 5 MiB in memory
		//	u.BufferProvider = s3manager.NewBufferedReadSeekerWriteToPool(5 * 1024 * 1024)
		u.LeavePartsOnError = true
	})

	return &s3u
}

// Upload uploads a file. blocking!
func (s3u *S3Upload) Upload() {
	// Upload the file to S3.
	u := s3manager.UploadInput{
		Bucket: aws.String(s3u.bucket),
		Key:    aws.String(s3u.key),
		Body:   s3u.ZP.Reader,
		// ContentEncoding: aws.String("gzip"),
		// StorageClass:    aws.String("STANDARD"),
	}
	fmt.Printf(" uploadInput: bucket: %s  Key: %s", *u.Bucket, *u.Key)
	result, err := s3u.uploader.Upload(&u)
	if err != nil {
		fmt.Fprintf(os.Stderr, "failed to upload file, %v", err)
	}
	fmt.Printf("file uploaded to, %s\n", result.Location)
}