upload.go

package main

import (
  "bufio"
	"log"
	"io"
	"os"
	"path"
	"strings"
	"sync"
	
	"launchpad.net/goamz/aws"
	"launchpad.net/goamz/s3"
)

func upload(b *s3.Bucket, match string) (err error) {
	name := path.Base(match)
	log.Println("uploading:", name)
	fi, err := os.Stat(match)
	if err != nil {
		return err
	}
	f, err := os.Open(match)
	if err != nil {
		return err
	}
	reader := bufio.NewReader(f)
	length := fi.Size()
	return b.PutReader(name, reader, length, "application/x-lzop", s3.Private)
}

type manifest map[string]string

func main() {
	f, err := os.OpenFile("manifest-md5.txt", os.O_RDONLY, 0666)
	if err != nil {
		log.Fatal("could not open manifest")
	}

	r := bufio.NewReader(f)

	m := manifest{}
	for {
		line, isPrefix, err := r.ReadLine()
		if isPrefix == true {
			log.Fatal("NYI")
		}
		if err==io.EOF {
			break
		}
		f := strings.Fields(string(line))
		if len(f)==2 {
			m[f[1]] = f[0]
		}
	}

	auth, err := aws.EnvAuth()
	if err != nil {
		log.Fatal(err)
	}

	s := s3.New(auth, aws.USEast)

	b := s.Bucket("twitter-unsorted")
	err = b.PutBucket(s3.Private)
	if err!=nil {
		log.Fatal(err)
	}

	uploadPaths := make(chan string, 1000)
	wg := sync.WaitGroup{}
	for i := 10; i>0; i-- {
		go func () {
			wg.Add(1)
			for path := range(uploadPaths) {
				err := upload(b, path)
				if err != nil {
					log.Println("ERROR:", err)
				}
			}
			wg.Done()
		}()
	}

	resp, err := b.List("", "/", "", 1000)
	if err != nil {
		log.Fatal(err)
	}

	key := func(name string) *s3.Key {
		for _, c := range resp.Contents {
			if name == c.Key {
				return &c
			}
		}
		return nil
	}

	for p, cs := range m {
		name := path.Base(p)
		k := key(name)
		if k == nil {
			uploadPaths <- p
		} else {
			etag := strings.Trim(k.ETag, `"`)
			if etag != cs {
				log.Printf("checksum mismatch '%s' vs '%s' for '%s'\n", etag, cs, name)
			}
		}
	}
	close(uploadPaths)
	wg.Wait()
}