cmd/swarm: progressbars initial version

api, cmd: fixes after rebase master

cmd/swarm: make progressbars show up by default

change for loop

cmd/swarm: add chunk count to progress bars

cmd/swarm: fix tests with new flag

cmd/swarm: remove number of chunks

cmd/swarm: add debug mode for progress bars

api/client/client.go

@@ -41,6 +41,7 @@ import (
 	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/ethersphere/swarm/api"
 	swarmhttp "github.com/ethersphere/swarm/api/http"
+	"github.com/ethersphere/swarm/chunk"
 	"github.com/ethersphere/swarm/spancontext"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/pborman/uuid"
@@ -660,6 +661,33 @@ func (c *Client) MultipartUpload(hash string, uploader Uploader, toPin bool) (st
 	return string(data), nil
 }
 
+// TagByHash queries the Swarm node for the state of the tag associated with the hash
+// this is used to poll progress while push-syncing
+func (c *Client) TagByHash(hash string) (*chunk.Tag, error) {
+	req, err := http.NewRequest("GET", c.Gateway+"/bzz-tag:/"+hash, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+	if res.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected HTTP status: %s", res.Status)
+	}
+	data, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	tag := &chunk.Tag{}
+	err = json.Unmarshal(data, tag)
+
+	return tag, err
+}
+
 // ErrNoFeedUpdatesFound is returned when Swarm cannot find updates of the given feed
 var ErrNoFeedUpdatesFound = errors.New("No updates found for this feed")
 

api/client/client_test.go

@@ -395,6 +395,32 @@ func TestClientMultipartUpload(t *testing.T) {
 	}
 }
 
+// TestClientQueryTagByHash tests that the correct reply is received in regards to a hash of an ongoing upload
+func TestClientQueryTagByHash(t *testing.T) {
+	srv := swarmhttp.NewTestSwarmServer(t, serverFunc, nil, nil)
+	defer srv.Close()
+
+	data := []byte("foo123")
+	client := NewClient(srv.URL)
+
+	hash, err := client.UploadRaw(bytes.NewReader(data), int64(len(data)), false, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tagg, err := client.TagByHash(hash)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// check the tag was created successfully
+	tag := srv.Tags.All()[0]
+	chunktesting.CheckTag(t, tag, 1, 1, 0, 1, 0, 1)
+
+	// check that the tag we got back from the API is also correct
+	chunktesting.CheckTag(t, tagg, 1, 1, 0, 1, 0, 1)
+}
+
 func newTestSigner() (*feed.GenericSigner, error) {
 	privKey, err := crypto.HexToECDSA("deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
 	if err != nil {

cmd/swarm/access_test.go

@@ -88,6 +88,7 @@ func testPassword(t *testing.T, cluster *testCluster) {
 		"--bzzapi",
 		cluster.Nodes[0].URL,
 		"up",
+		"--no-track",
 		"--encrypt",
 		dataFilename)
 	_, matches := up.ExpectRegexp(hashRegexp)
@@ -240,6 +241,7 @@ func testPK(t *testing.T, cluster *testCluster) {
 		"--bzzapi",
 		cluster.Nodes[0].URL,
 		"up",
+		"--no-track",
 		"--encrypt",
 		dataFilename)
 	_, matches := up.ExpectRegexp(hashRegexp)
@@ -392,6 +394,7 @@ func testACT(t *testing.T, cluster *testCluster, bogusEntries int) {
 		"--bzzapi",
 		cluster.Nodes[0].URL,
 		"up",
+		"--no-track",
 		"--encrypt",
 		dataFilename)
 	_, matches := up.ExpectRegexp(hashRegexp)

cmd/swarm/export_test.go

@@ -64,7 +64,7 @@ func TestCLISwarmExportImport(t *testing.T) {
 	defer os.Remove(fileName)
 
 	// upload the file with 'swarm up' and expect a hash
-	up := runSwarm(t, "--bzzapi", cluster.Nodes[0].URL, "up", fileName)
+	up := runSwarm(t, "--bzzapi", cluster.Nodes[0].URL, "up", "--no-track", fileName)
 	_, matches := up.ExpectRegexp(`[a-f\d]{64}`)
 	up.ExpectExit()
 	hash := matches[0]

cmd/swarm/flags.go

@@ -233,4 +233,8 @@ var (
 		Name:  "enable-pinning",
 		Usage: "Use this flag to enable the pinning feature",
 	}
+	SwarmNoTrackUploadFlag = cli.BoolFlag{
+		Name:  "no-track",
+		Usage: "Use this flag to disable tracking of the upload progress through the CLI (gives back a machine-readable content addressed hash)",
+	}
 )

cmd/swarm/fs_test.go

@@ -222,6 +222,7 @@ func doUploadEmptyDir(t *testing.T, node *testNode) string {
 		"--bzzapi", node.URL,
 		"--recursive",
 		"up",
+		"--no-track",
 		tmpDir}
 
 	log.Info("swarmfs cli test: uploading dir with 'swarm up'")

cmd/swarm/manifest_test.go

@@ -100,11 +100,12 @@ func testManifestChange(t *testing.T, encrypt bool) {
 		"--defaultpath",
 		indexDataFilename,
 		"up",
-		origDir,
+		"--no-track",
 	}
 	if encrypt {
 		args = append(args, "--encrypt")
 	}
+	args = append(args, origDir)
 
 	origManifestHash := runSwarmExpectHash(t, args...)
 
@@ -125,6 +126,7 @@ func testManifestChange(t *testing.T, encrypt bool) {
 			"--bzzapi",
 			srv.URL,
 			"up",
+			"--no-track",
 			humansDataFilename,
 		)
 
@@ -180,6 +182,7 @@ func testManifestChange(t *testing.T, encrypt bool) {
 			"--bzzapi",
 			srv.URL,
 			"up",
+			"--no-track",
 			robotsDataFilename,
 		)
 
@@ -240,6 +243,7 @@ func testManifestChange(t *testing.T, encrypt bool) {
 			"--bzzapi",
 			srv.URL,
 			"up",
+			"--no-track",
 			indexDataFilename,
 		)
 
@@ -298,6 +302,7 @@ func testManifestChange(t *testing.T, encrypt bool) {
 			"--bzzapi",
 			srv.URL,
 			"up",
+			"--no-track",
 			robotsDataFilename,
 		)
 
@@ -464,12 +469,12 @@ func testNestedDefaultEntryUpdate(t *testing.T, encrypt bool) {
 		"--defaultpath",
 		indexDataFilename,
 		"up",
-		origDir,
+		"--no-track",
 	}
 	if encrypt {
 		args = append(args, "--encrypt")
 	}
-
+	args = append(args, origDir)
 	origManifestHash := runSwarmExpectHash(t, args...)
 
 	checkHashLength(t, origManifestHash, encrypt)
@@ -487,6 +492,7 @@ func testNestedDefaultEntryUpdate(t *testing.T, encrypt bool) {
 		"--bzzapi",
 		srv.URL,
 		"up",
+		"--no-track",
 		newIndexDataFilename,
 	)
 

cmd/swarm/upload.go

@@ -28,23 +28,41 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethersphere/swarm/api/client"
 	swarm "github.com/ethersphere/swarm/api/client"
+	"github.com/ethersphere/swarm/chunk"
+	"github.com/vbauerster/mpb"
+	"github.com/vbauerster/mpb/decor"
 
 	"github.com/ethereum/go-ethereum/cmd/utils"
 	"gopkg.in/urfave/cli.v1"
 )
 
-var upCommand = cli.Command{
-	Action:             upload,
-	CustomHelpTemplate: helpTemplate,
-	Name:               "up",
-	Usage:              "uploads a file or directory to swarm using the HTTP API",
-	ArgsUsage:          "<file>",
-	Flags:              []cli.Flag{SwarmEncryptedFlag, SwarmPinFlag},
-	Description:        "uploads a file or directory to swarm using the HTTP API and prints the root hash",
-}
+var (
+	upCommand = cli.Command{
+		Action:             upload,
+		CustomHelpTemplate: helpTemplate,
+		Name:               "up",
+		Usage:              "uploads a file or directory to swarm using the HTTP API",
+		ArgsUsage:          "<file>",
+		Flags:              []cli.Flag{SwarmEncryptedFlag, SwarmPinFlag, SwarmNoTrackUploadFlag},
+		Description:        "uploads a file or directory to swarm using the HTTP API and prints the root hash",
+	}
+
+	pollDelay   = 200 * time.Millisecond
+	chunkStates = []struct {
+		name  string
+		state chunk.State
+	}{
+		{"Split", chunk.StateSplit},
+		{"Stored", chunk.StateStored},
+		{"Sent", chunk.StateSent},
+		{"Synced", chunk.StateSynced},
+	}
+)
 
 func upload(ctx *cli.Context) {
 	args := ctx.Args()
@@ -55,12 +73,17 @@ func upload(ctx *cli.Context) {
 		defaultPath     = ctx.GlobalString(SwarmUploadDefaultPath.Name)
 		fromStdin       = ctx.GlobalBool(SwarmUpFromStdinFlag.Name)
 		mimeType        = ctx.GlobalString(SwarmUploadMimeType.Name)
+		debug           = ctx.GlobalBool("debug")
 		client          = swarm.NewClient(bzzapi)
 		toEncrypt       = ctx.Bool(SwarmEncryptedFlag.Name)
 		toPin           = ctx.Bool(SwarmPinFlag.Name)
+		notrack         = ctx.Bool(SwarmNoTrackUploadFlag.Name)
 		autoDefaultPath = false
 		file            string
 	)
+	if !debug {
+		chunkStates = chunkStates[3:] // just poll Synced state
+	}
 	if autoDefaultPathString := os.Getenv(SwarmAutoDefaultPath); autoDefaultPathString != "" {
 		b, err := strconv.ParseBool(autoDefaultPathString)
 		if err != nil {
@@ -155,7 +178,118 @@ func upload(ctx *cli.Context) {
 	if err != nil {
 		utils.Fatalf("Upload failed: %s", err)
 	}
-	fmt.Println(hash)
+
+	// dont show the progress bar (machine readable output)
+	if notrack {
+		fmt.Println(hash)
+		return
+	}
+
+	// this section renders the cli UI for showing the progress bars
+	tag, err := client.TagByHash(hash)
+	if err != nil {
+		utils.Fatalf("failed to get tag data for hash: %v", err)
+	}
+	fmt.Println("Swarm Hash:", hash)
+	fmt.Println("Tag UID:", tag.Uid)
+	// check if the user uploaded something that was already completely stored
+	// in the local store (otherwise we hang forever because there's nothing to sync)
+	// as the chunks are already supposed to be synced
+	seen, total, err := tag.Status(chunk.StateSeen)
+	if total-seen > 0 {
+		fmt.Println("Upload status:")
+		bars := createTagBars(tag, debug)
+		pollTag(client, tag, bars, debug)
+	}
+
+	fmt.Println("Done! took", time.Since(tag.StartedAt))
+	fmt.Println("Your Swarm hash should now be retrievable from other nodes!")
+}
+
+func pollTag(client *client.Client, tag *chunk.Tag, bars map[string]*mpb.Bar, debug bool) {
+	oldTag := *tag
+	lastTime := time.Now()
+
+	for {
+		time.Sleep(pollDelay)
+		newTag, err := client.TagByHash(tag.Address.String())
+		if err != nil {
+			utils.Fatalf("had an error polling the tag for address %s, err %v", tag.Address.String(), err)
+		}
+		done := true
+		for _, state := range chunkStates {
+			// calculate the difference that we need to increment for each bar
+			count, _, err := oldTag.Status(state.state)
+			if err != nil {
+				utils.Fatalf("error while getting tag status: %v", err)
+			}
+			newCount, total, err := newTag.Status(state.state)
+			if err != nil {
+				utils.Fatalf("error while getting tag status: %v", err)
+			}
+			d := int(newCount - count)
+			if newCount != total {
+				done = false
+			}
+			bars[state.name].SetTotal(total, done)
+			bars[state.name].IncrBy(d, time.Since(lastTime))
+		}
+		if done {
+			return
+		}
+
+		oldTag = *newTag
+		lastTime = time.Now()
+	}
+}
+
+func createTagBars(tag *chunk.Tag, debug bool) map[string]*mpb.Bar {
+	p := mpb.New(mpb.WithWidth(64))
+	bars := make(map[string]*mpb.Bar)
+	for _, state := range chunkStates {
+		count, total, err := tag.Status(state.state)
+		if err != nil {
+			utils.Fatalf("could not get tag status: %v", err)
+		}
+		title := state.name
+		var barElement *mpb.Bar
+		width := 10
+		if debug {
+			barElement = p.AddBar(total,
+				mpb.PrependDecorators(
+					// align the elements with a constant size (10 chars)
+					decor.Name(title, decor.WC{W: width, C: decor.DidentRight}),
+					// add unit counts
+					decor.CountersNoUnit("%d / %d", decor.WCSyncSpace),
+					// replace ETA decorator with "done" message, OnComplete event
+					decor.OnComplete(
+						// ETA decorator with ewma age of 60, and width reservation of 4
+						decor.EwmaETA(decor.ET_STYLE_GO, 60, decor.WC{W: 6}), "done",
+					),
+				),
+				mpb.AppendDecorators(decor.Percentage()),
+			)
+		} else {
+			title = fmt.Sprintf("Syncing %d chunks", total)
+			width = len(title) + 3
+			barElement = p.AddBar(total,
+				mpb.PrependDecorators(
+					// align the elements with a constant size (10 chars)
+					decor.Name(title, decor.WC{W: width, C: decor.DidentRight}),
+					// replace ETA decorator with "done" message, OnComplete event
+					decor.OnComplete(
+						// ETA decorator with ewma age of 60, and width reservation of 4
+						decor.EwmaETA(decor.ET_STYLE_GO, 60, decor.WC{W: 6}), "done",
+					),
+				),
+				mpb.AppendDecorators(decor.Percentage()),
+			)
+		}
+		// increment the bar with the initial value from the tag
+		barElement.IncrBy(int(count))
+		bars[state.name] = barElement
+	}
+	return bars
 }
 
 // Expands a file path