func (a *Archive) worker(wg *sync.WaitGroup, pending chan *s3.Object, objects chan *replay.Object) {
	for o := range pending {
		buff := &aws.WriteAtBuffer{}
		n, err := a.downloader.Download(buff, &s3.GetObjectInput{
			Bucket: a.bucket,
			Key:    o.Key,
		})
		if err != nil {
			a.log.WithError(err).Errorln("download error")
			pending <- o
			break
		}
		a.log.WithFields(logrus.Fields{
			"n":   n,
			"key": o.Key,
		}).Debugln("download complete")
		objects <- &replay.Object{
			Data:   buff.Bytes(),
			Object: o,
		}
	}
	wg.Done()
}
worker manages downloading pending s3 objects
package s3
import (
	"io"
	"s3-kinesis-replay/replay"
	"s3-kinesis-replay/validate"
	"sync"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
	"github.com/aws/aws-sdk-go/service/s3/s3manager"
	"github.com/sirupsen/logrus"
)
type Archive struct {
	bucket      *string
	concurrency int
	client      s3iface.S3API
	downloader  Downloader
	log         logrus.FieldLogger
	prefix      *string
	startAfter  *string
	stopAt      *string
	wg          *sync.WaitGroup
}
func NewArchive(c *ArchiveConfig) (*Archive, error) {
	err := validate.V.Struct(c)
	if err != nil {
		return nil, err
	}
	a := &Archive{
		bucket:      aws.String(c.Bucket),
		client:      c.Client,
		concurrency: c.Concurrency,
		downloader:  c.Downloader,
		log:         c.Log,
		wg:          &sync.WaitGroup{},
	}
	if c.Prefix != "" {
		a.prefix = aws.String(c.Prefix)
	}
	if c.StartAfter != "" {
		a.startAfter = aws.String(c.StartAfter)
	}
	if c.StopAt != "" {
		a.stopAt = aws.String(c.StopAt)
	}
	return a, nil
}
func (a *Archive) Scan() chan *replay.Object {
	pending := make(chan *s3.Object, 1000)
	objects := make(chan *replay.Object, 1000)
	for i := 0; i < a.concurrency; i++ {
		a.wg.Add(1)
		go a.worker(a.wg, pending, objects)
	}
	go a.scan(pending)
	go func() {
		a.wg.Wait()
		close(objects)
	}()
	return objects
}
func (a *Archive) scan(pending chan *s3.Object) {
	params := &s3.ListObjectsV2Input{
		Bucket:     a.bucket,
		Prefix:     a.prefix,
		StartAfter: a.startAfter,
	}
	err := a.client.ListObjectsV2Pages(params, func(output *s3.ListObjectsV2Output, more bool) bool {
		for _, o := range output.Contents {
			if a.stopAt != nil && *a.stopAt == *o.Key {
				a.log.WithField("key", *o.Key).Infoln("stopping at stop key")
				return false
			}
			a.log.WithField("key", *o.Key).Debugln("queueing s3 key for download")
			pending <- o
		}
		return true
	})
	if err != nil {
		a.log.WithError(err).Errorln("scan:error")
	} else {
		a.log.Infoln("scan complete")
	}
	close(pending)
}