Merge pull request #1600 from 0chain/feat/repair-progress

Repair progress

wasmsdk/allocation.go

@@ -248,7 +248,7 @@ func updateAllocationWithRepair(allocationID string,
 	}
 
 	wg := &sync.WaitGroup{}
-	statusBar := &StatusBar{wg: wg, isRepair: true}
+	statusBar := &StatusBar{wg: wg, isRepair: true, totalBytesMap: make(map[string]int)}
 	wg.Add(1)
 
 	alloc, hash, isRepairRequired, err := allocationObj.UpdateWithStatus(size, extend, uint64(lock), addBlobberId, addBlobberAuthTicket, removeBlobberId, false, &sdk.FileOptionsParameters{}, statusBar)
@@ -520,7 +520,7 @@ func allocationRepair(allocationID, remotePath string) error {
 	}
 	sdk.SetWasm()
 	wg := &sync.WaitGroup{}
-	statusBar := &StatusBar{wg: wg, isRepair: true}
+	statusBar := &StatusBar{wg: wg, isRepair: true, totalBytesMap: make(map[string]int)}
 	wg.Add(1)
 
 	err = allocationObj.StartRepair("/tmp", remotePath, statusBar)

wasmsdk/blobber.go

@@ -467,7 +467,7 @@ func multiDownload(allocationID, jsonMultiDownloadOptions, authTicket, callbackF
 		fileName := strings.Replace(path.Base(option.RemotePath), "/", "-", -1)
 		localPath := allocationID + "_" + fileName
 		option.LocalPath = localPath
-		statusBar := &StatusBar{wg: wg}
+		statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 		allStatusBar[ind] = statusBar
 		if useCallback {
 			callback := js.Global().Get(callbackFuncName)
@@ -729,7 +729,7 @@ func multiUpload(jsonBulkUploadOptions string) (MultiUploadResult, error) {
 	operationRequests := make([]sdk.OperationRequest, n)
 	for idx, option := range options {
 		wg := &sync.WaitGroup{}
-		statusBar := &StatusBar{wg: wg}
+		statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 		callbackFuncName := option.CallbackFuncName
 		if callbackFuncName != "" {
 			callback := js.Global().Get(callbackFuncName)
@@ -834,7 +834,7 @@ func uploadWithJsFuncs(allocationID, remotePath string, readChunkFuncName string
 	}
 
 	wg := &sync.WaitGroup{}
-	statusBar := &StatusBar{wg: wg}
+	statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	if callbackFuncName != "" {
 		callback := js.Global().Get(callbackFuncName)
 		statusBar.callback = func(totalBytes, completedBytes int, filename, objURL, err string) {
@@ -930,7 +930,7 @@ func upload(allocationID, remotePath string, fileBytes, thumbnailBytes []byte, w
 	}
 
 	wg := &sync.WaitGroup{}
-	statusBar := &StatusBar{wg: wg}
+	statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	wg.Add(1)
 
 	fileReader := bytes.NewReader(fileBytes)
@@ -1013,7 +1013,7 @@ func downloadBlocks(allocId string, remotePath, authTicket, lookupHash string, s
 
 	var (
 		wg        = &sync.WaitGroup{}
-		statusBar = &StatusBar{wg: wg}
+		statusBar = &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	)
 
 	fileName := strings.Replace(path.Base(remotePath), "/", "-", -1)
@@ -1064,7 +1064,7 @@ func getBlobbers(stakable bool) ([]*sdk.Blobber, error) {
 // repairAllocation repair the allocation
 // Allocation repair is a process to repair the allocation files on its blobbers by re-uploading the missing blocks.
 //   - allocationID : allocation ID of the file
-func repairAllocation(allocationID string) error {
+func repairAllocation(allocationID, callbackFuncName string) error {
 	alloc, err := getAllocation(allocationID)
 	if err != nil {
 		return err
@@ -1073,16 +1073,21 @@ func repairAllocation(allocationID string) error {
 	if err != nil {
 		return err
 	}
-	statusBar := sdk.NewRepairBar(allocationID)
-	if statusBar == nil {
-		return errors.New("repair already in progress")
+	wg := &sync.WaitGroup{}
+	statusBar := &StatusBar{wg: wg, isRepair: true, totalBytesMap: make(map[string]int)}
+	wg.Add(1)
+	if callbackFuncName != "" {
+		callback := js.Global().Get(callbackFuncName)
+		statusBar.callback = func(totalBytes, completedBytes int, filename, objURL, err string) {
+			callback.Invoke(totalBytes, completedBytes, filename, objURL, err)
+		}
 	}
 	err = alloc.RepairAlloc(statusBar)
 	if err != nil {
 		return err
 	}
-	statusBar.Wait()
-	return statusBar.CheckError()
+	wg.Wait()
+	return statusBar.err
 }
 
 // checkAllocStatus check the status of the allocation, either it is ok, needs repair or broken
@@ -1166,7 +1171,7 @@ func downloadDirectory(allocationID, remotePath, authticket, callbackFuncName st
 	}
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
-	statusBar := &StatusBar{wg: wg}
+	statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	if callbackFuncName != "" {
 		callback := js.Global().Get(callbackFuncName)
 		statusBar.callback = func(totalBytes, completedBytes int, filename, objURL, err string) {

wasmsdk/player_stream.go

@@ -69,7 +69,7 @@ func (p *StreamPlayer) Stop() {
 
 func (p *StreamPlayer) download(it sdk.PlaylistFile) {
 	wg := &sync.WaitGroup{}
-	statusBar := &StatusBar{wg: wg}
+	statusBar := &StatusBar{wg: wg, totalBytesMap: make(map[string]int)}
 	wg.Add(1)
 
 	fileName := it.Name

wasmsdk/statusbar.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/0chain/gosdk/core/sys"
+	"github.com/0chain/gosdk/zboxcore/sdk"
 	"gopkg.in/cheggaaa/pb.v1"
 )
 
@@ -24,6 +25,7 @@ type StatusBar struct {
 	localPath      string
 	callback       func(totalBytes int, completedBytes int, fileName, objURL, err string)
 	isRepair       bool
+	totalBytesMap  map[string]int
 }
 
 var jsCallbackMutex sync.Mutex
@@ -37,9 +39,15 @@ func (s *StatusBar) Started(allocationID, filePath string, op int, totalBytes in
 	fileName := path.Base(filePath)
 	s.totalBytes = totalBytes
 	if s.callback != nil {
-		jsCallbackMutex.Lock()
-		defer jsCallbackMutex.Unlock()
-		s.callback(s.totalBytes, s.completedBytes, fileName, "", "")
+		if !s.isRepair || op == sdk.OpUpload || op == sdk.OpUpdate {
+			if s.isRepair {
+				fileName = filePath
+			}
+			jsCallbackMutex.Lock()
+			defer jsCallbackMutex.Unlock()
+			s.totalBytesMap[filePath] = totalBytes
+			s.callback(totalBytes, s.completedBytes, fileName, "", "")
+		}
 	}
 }
 
@@ -49,11 +57,15 @@ func (s *StatusBar) InProgress(allocationID, filePath string, op int, completedB
 		s.b.Set(completedBytes)
 	}
 	fileName := path.Base(filePath)
-	s.completedBytes = completedBytes
 	if s.callback != nil {
-		jsCallbackMutex.Lock()
-		defer jsCallbackMutex.Unlock()
-		s.callback(s.totalBytes, s.completedBytes, fileName, "", "")
+		if !s.isRepair || op == sdk.OpUpload || op == sdk.OpUpdate {
+			if s.isRepair {
+				fileName = filePath
+			}
+			jsCallbackMutex.Lock()
+			defer jsCallbackMutex.Unlock()
+			s.callback(s.totalBytesMap[filePath], completedBytes, fileName, "", "")
+		}
 	}
 }
 
@@ -64,16 +76,22 @@ func (s *StatusBar) Completed(allocationID, filePath string, filename string, mi
 	}
 	s.success = true
 
-	s.completedBytes = s.totalBytes
 	if s.localPath != "" {
 		fs, _ := sys.Files.Open(s.localPath)
 		mf, _ := fs.(*sys.MemFile)
 		s.objURL = CreateObjectURL(mf.Buffer, mimetype)
 	}
 	if s.callback != nil {
-		jsCallbackMutex.Lock()
-		defer jsCallbackMutex.Unlock()
-		s.callback(s.totalBytes, s.completedBytes, filename, s.objURL, "")
+		if !s.isRepair || op == sdk.OpUpload || op == sdk.OpUpdate {
+			if s.isRepair {
+				filename = filePath
+			}
+			jsCallbackMutex.Lock()
+			defer jsCallbackMutex.Unlock()
+			totalBytes := s.totalBytesMap[filePath]
+			delete(s.totalBytesMap, filePath)
+			s.callback(totalBytes, totalBytes, filename, s.objURL, "")
+		}
 	}
 	if !s.isRepair {
 		defer s.wg.Done()
@@ -96,9 +114,14 @@ func (s *StatusBar) Error(allocationID string, filePath string, op int, err erro
 	fileName := path.Base(filePath)
 	PrintError("Error in file operation." + err.Error())
 	if s.callback != nil {
-		jsCallbackMutex.Lock()
-		defer jsCallbackMutex.Unlock()
-		s.callback(s.totalBytes, s.completedBytes, fileName, "", err.Error())
+		if !s.isRepair || op == sdk.OpUpload || op == sdk.OpUpdate {
+			if s.isRepair {
+				fileName = filePath
+			}
+			jsCallbackMutex.Lock()
+			defer jsCallbackMutex.Unlock()
+			s.callback(s.totalBytesMap[filePath], s.completedBytes, fileName, "", err.Error())
+		}
 	}
 	if !s.isRepair {
 		s.wg.Done()