Merge pull request #231 from ipfs/fix/224-pin-progress

Fix #224: Better handling of progress updates when proxy-adding file

ipfsconn/ipfshttp/ipfshttp.go

@@ -87,8 +87,9 @@ type ipfsRepoStatResp struct {
 }
 
 type ipfsAddResp struct {
-	Name string
-	Hash string
+	Name  string
+	Hash  string
+	Bytes uint64
 }
 
 // NewConnector creates the component and leaves it ready to be started
@@ -361,9 +362,6 @@ func (ipfs *Connector) addHandler(w http.ResponseWriter, r *http.Request) {
 	// Cluster will decide where to pin based on metrics and current
 	// allocations.
 	q.Set("pin", "false")
-	// do not send progress updates. They have no use since we need to
-	// wait until the end to tell cluster to pin.
-	q.Set("progress", "false")
 	r.URL.RawQuery = q.Encode()
 
 	res, err := ipfs.proxyRequest(r)
@@ -405,6 +403,10 @@ func (ipfs *Connector) addHandler(w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "error decoding response: "+err.Error(), 502)
 			return
 		}
+		if addResp.Bytes != 0 {
+			// This is a progress notification, so we ignore it
+			continue
+		}
 		ipfsAddResps = append(ipfsAddResps, addResp)
 	}
 

ipfsconn/ipfshttp/ipfshttp_test.go

@@ -391,11 +391,24 @@ func TestProxyAdd(t *testing.T) {
 		}
 
 		var hash ipfsAddResp
-		resBytes, _ := ioutil.ReadAll(res.Body)
-		err = json.Unmarshal(resBytes, &hash)
-		if err != nil {
-			t.Fatal(err)
+
+		// We might return a progress notification, so we do it
+		// like this to ignore it easily
+		dec := json.NewDecoder(res.Body)
+		for dec.More() {
+			var resp ipfsAddResp
+			err := dec.Decode(&resp)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if resp.Bytes != 0 {
+				continue
+			} else {
+				hash = resp
+			}
 		}
+
 		if hash.Hash != test.TestCid3 {
 			t.Logf("%+v", hash)
 			t.Error("expected TestCid1 as it is hardcoded in ipfs mock")
@@ -431,37 +444,37 @@ func TestDecideRecursivePins(t *testing.T) {
 
 	tcs := []testcases{
 		{
-			[]ipfsAddResp{{"a", "cida"}},
+			[]ipfsAddResp{{"a", "cida", 0}},
 			url.Values{},
 			[]string{"cida"},
 		},
 		{
-			[]ipfsAddResp{{"a/b", "cidb"}, {"a", "cida"}},
+			[]ipfsAddResp{{"a/b", "cidb", 0}, {"a", "cida", 0}},
 			url.Values{},
 			[]string{"cida"},
 		},
 		{
-			[]ipfsAddResp{{"a/b", "cidb"}, {"c", "cidc"}, {"a", "cida"}},
+			[]ipfsAddResp{{"a/b", "cidb", 0}, {"c", "cidc", 0}, {"a", "cida", 0}},
 			url.Values{},
 			[]string{"cidc", "cida"},
 		},
 		{
-			[]ipfsAddResp{{"/a", "cida"}},
+			[]ipfsAddResp{{"/a", "cida", 0}},
 			url.Values{},
 			[]string{"cida"},
 		},
 		{
-			[]ipfsAddResp{{"a/b/c/d", "cidd"}},
+			[]ipfsAddResp{{"a/b/c/d", "cidd", 0}},
 			url.Values{},
 			[]string{"cidd"},
 		},
 		{
-			[]ipfsAddResp{{"a", "cida"}, {"b", "cidb"}, {"c", "cidc"}, {"d", "cidd"}},
+			[]ipfsAddResp{{"a", "cida", 0}, {"b", "cidb", 0}, {"c", "cidc", 0}, {"d", "cidd", 0}},
 			url.Values{},
 			[]string{"cida", "cidb", "cidc", "cidd"},
 		},
 		{
-			[]ipfsAddResp{{"a", "cida"}, {"b", "cidb"}, {"", "cidwrap"}},
+			[]ipfsAddResp{{"a", "cida", 0}, {"b", "cidb", 0}, {"", "cidwrap", 0}},
 			url.Values{
 				"wrap-in-directory": []string{"true"},
 			},
@@ -469,7 +482,7 @@ func TestDecideRecursivePins(t *testing.T) {
 		},
 		{
 
-			[]ipfsAddResp{{"b", ""}, {"a", "cida"}},
+			[]ipfsAddResp{{"b", "", 0}, {"a", "cida", 0}},
 			url.Values{},
 			[]string{"cida"},
 		},

test/ipfs_mock.go

@@ -58,8 +58,9 @@ type mockConfigResp struct {
 }
 
 type mockAddResp struct {
-	Name string
-	Hash string
+	Name  string
+	Hash  string
+	Bytes uint64
 }
 
 // NewIpfsMock returns a new mock.
@@ -106,6 +107,17 @@ func (m *IpfsMock) handler(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
+		query := r.URL.Query()
+		progress, ok := query["progress"]
+		if ok && len(progress) > 0 && progress[0] != "false" {
+			progressResp := mockAddResp{
+				Name:  fheader.Filename,
+				Bytes: 4,
+			}
+			j, _ := json.Marshal(progressResp)
+			w.Write(j)
+		}
+
 		resp := mockAddResp{
 			Name: fheader.Filename,
 			Hash: TestCid3,