Add progress updating to the picasa client.

clients/picasa/picasa.go

@@ -25,17 +25,17 @@ import (
 var (
 	authHTTPClient    *http.Client
 	cachingHTTPClient *http.Client
+	clientFlags       = util.NewFlags() // TODO: respect clientFlags.Concurrency()
 	ds                *dataset.Dataset
 	tokenFlag         = flag.String("token", "", "OAuth2 bearer token to authenticate with (required)")
-	progressFileFlag  = flag.String("progress-file", "", "file for logging progress")
 	start             time.Time
+	progress          chan int
 )
 
 type shapeMap map[string][]Shape
 
 func main() {
 	flag.Usage = picasaUsage
-	dsFlags := dataset.NewFlags()
 	flag.Parse()
 	cachingHTTPClient = util.CachingHttpClient()
 
@@ -47,19 +47,24 @@ func main() {
 	util.MaybeStartCPUProfile()
 	defer util.StopCPUProfile()
 
-	ds = dsFlags.CreateDataset()
+	ds = clientFlags.CreateDataset()
 	if ds == nil {
 		flag.Usage()
 		os.Exit(1)
 	}
 	defer ds.Store().Close()
 
+	if err := clientFlags.CreateProgressFile(); err != nil {
+		fmt.Println(err)
+	} else {
+		defer clientFlags.CloseProgressFile()
+	}
+
 	token := oauth2.Token{AccessToken: *tokenFlag}
 	authHTTPClient = oauth2.NewClient(oauth2.NoContext, oauth2.StaticTokenSource(&token))
 
 	start = time.Now()
 	user := getUser()
-	printStats(user)
 
 	userRef := types.WriteValue(user, ds.Store())
 	fmt.Printf("userRef: %s\n", userRef)
@@ -82,7 +87,26 @@ func picasaUsage() {
 func getUser() User {
 	alj := AlbumListJSON{}
 	callPicasaAPI(authHTTPClient, "user/default?alt=json", &alj)
-	fmt.Printf("Found %d albums\n", len(alj.Feed.Entry))
+
+	numPhotos := 0
+	for _, entry := range alj.Feed.Entry {
+		numPhotos += entry.NumPhotos.V
+	}
+	fmt.Printf("Found %d albums with %d photos\n", len(alj.Feed.Entry), numPhotos)
+
+	progress = make(chan int, numPhotos)
+	go func() {
+		added := 0
+		lastUpdate := time.Now()
+		for i := 0; i < numPhotos; i++ {
+			added += <-progress
+			if now := time.Now(); now.Sub(lastUpdate)/time.Millisecond >= 200 {
+				clientFlags.UpdateProgress(float32(added) / float32(numPhotos))
+				lastUpdate = now
+			}
+		}
+	}()
+
 	albums := ListOfRefOfAlbumDef{}
 	user := NewUser().
 		SetId(alj.Feed.UserID.V).
@@ -113,6 +137,7 @@ func getUser() User {
 
 	wg.Wait()
 
+	clientFlags.UpdateProgress(1.0)
 	return user.SetAlbums(albums.New())
 }
 
@@ -188,6 +213,7 @@ func getRemotePhotos(albumId string, numPhotos int, shapes shapeMap) SetOfRefOfR
 				mu.Lock()
 				remotePhotos[types.WriteValue(p, ds.Store())] = true
 				mu.Unlock()
+				progress <- 1
 			}
 		}()
 	}
@@ -201,17 +227,6 @@ func parsePoint(s string) (x, y float32) {
 	return float32(atoi(pair[0])), float32(atoi(pair[1]))
 }
 
-func printStats(user User) {
-	numPhotos := uint64(0)
-	albums := user.Albums()
-	albums.IterAll(func(album RefOfAlbum, _ uint64) {
-		// TODO: Sucks to deref album just to get num photos
-		numPhotos = numPhotos + album.TargetValue(ds.Store()).Photos().Len()
-	})
-
-	fmt.Printf("Imported %d album(s), %d photo(s), time: %.2f\n", albums.Len(), numPhotos, time.Now().Sub(start).Seconds())
-}
-
 func callPicasaAPI(client *http.Client, path string, response interface{}) {
 	rc := callPicasaURL(client, "https://picasaweb.google.com/data/feed/api/"+path)
 	defer rc.Close()

clients/util/client_flags.go

@@ -0,0 +1,63 @@
+package util
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/attic-labs/noms/d"
+	"github.com/attic-labs/noms/dataset"
+)
+
+type ClientFlags struct {
+	dsFlags          dataset.DatasetFlags
+	concurrencyFlag  *int
+	progressFileFlag *string
+	progressFile     *os.File
+}
+
+func NewFlags() ClientFlags {
+	return NewFlagsWithPrefix("")
+}
+
+func NewFlagsWithPrefix(prefix string) ClientFlags {
+	return ClientFlags{
+		dataset.NewFlagsWithPrefix(prefix),
+		flag.Int(prefix+"concurrency", 100, "degree of concurrency"),
+		flag.String(prefix+"progress-file", "", "file for logging progress"),
+		nil,
+	}
+}
+
+func (cf *ClientFlags) CreateDataset() *dataset.Dataset {
+	return cf.dsFlags.CreateDataset()
+}
+
+func (cf *ClientFlags) Concurrency() int {
+	return *cf.concurrencyFlag
+}
+
+func (cf *ClientFlags) CreateProgressFile() error {
+	if *cf.progressFileFlag == "" {
+		return fmt.Errorf("No progress file specified")
+	}
+	pf, err := os.Create(*cf.progressFileFlag)
+	if err != nil {
+		return fmt.Errorf("Unable to create progress status file: %s, err: %s", *cf.progressFileFlag, err)
+	}
+	cf.progressFile = pf
+	cf.UpdateProgress(0.0)
+	return err
+}
+
+func (cf *ClientFlags) CloseProgressFile() {
+	d.Chk.NotNil(cf.progressFile, "Progress file was never created")
+	cf.progressFile.Close()
+}
+
+func (cf *ClientFlags) UpdateProgress(pct float32) {
+	d.Chk.True(pct >= 0.0 && pct <= 1.0, "%.6f is not a valid percentage, must be in range [0, 1]", pct)
+	if cf.progressFile != nil {
+		cf.progressFile.WriteString(fmt.Sprintf("%.6f\n", pct))
+	}
+}

dataset/dataset.go

@@ -107,23 +107,23 @@ func (ds *Dataset) SetNewHead(newHeadRef ref.Ref) (Dataset, error) {
 	return ds.CommitWithParents(commit.Value(), commit.Parents())
 }
 
-type datasetFlags struct {
+type DatasetFlags struct {
 	datas.Flags
 	datasetID *string
 }
 
-func NewFlags() datasetFlags {
+func NewFlags() DatasetFlags {
 	return NewFlagsWithPrefix("")
 }
 
-func NewFlagsWithPrefix(prefix string) datasetFlags {
-	return datasetFlags{
+func NewFlagsWithPrefix(prefix string) DatasetFlags {
+	return DatasetFlags{
 		datas.NewFlagsWithPrefix(prefix),
 		flag.String(prefix+"ds", "", "dataset id to store data for"),
 	}
 }
 
-func (f datasetFlags) CreateDataset() *Dataset {
+func (f DatasetFlags) CreateDataset() *Dataset {
 	if *f.datasetID == "" {
 		return nil
 	}