migrate volume sync to gRpc

1 files changed, 21 insertions, 14 deletions

diff --git a/weed/operation/sync_volume.go b/weed/operation/sync_volume.go
index 390d5b504..662184656 100644
--- a/weed/operation/sync_volume.go
+++ b/weed/operation/sync_volume.go
@@ -2,7 +2,6 @@ package operation
 
 import (
 	"context"
-	"net/url"
 
 	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
 	. "github.com/chrislusf/seaweedfs/weed/storage/types"
@@ -21,18 +20,26 @@ func GetVolumeSyncStatus(server string, vid uint32) (resp *volume_server_pb.Volu
 	return
 }
 
-func GetVolumeIdxEntries(server string, vid string, eachEntryFn func(key NeedleId, offset Offset, size uint32)) error {
-	values := make(url.Values)
-	values.Add("volume", vid)
-	line := make([]byte, NeedleEntrySize)
-	err := util.GetBufferStream("http://"+server+"/admin/sync/index", values, line, func(bytes []byte) {
-		key := BytesToNeedleId(line[:NeedleIdSize])
-		offset := BytesToOffset(line[NeedleIdSize : NeedleIdSize+OffsetSize])
-		size := util.BytesToUint32(line[NeedleIdSize+OffsetSize : NeedleIdSize+OffsetSize+SizeSize])
-		eachEntryFn(key, offset, size)
+func GetVolumeIdxEntries(server string, vid uint32, eachEntryFn func(key NeedleId, offset Offset, size uint32)) error {
+
+	return WithVolumeServerClient(server, func(client volume_server_pb.VolumeServerClient) error {
+		resp, err := client.VolumeSyncIndex(context.Background(), &volume_server_pb.VolumeSyncIndexRequest{
+			VolumdId: vid,
+		})
+		if err != nil {
+			return err
+		}
+
+		dataSize := len(resp.IndexFileContent)
+
+		for idx := 0; idx+NeedleEntrySize <= dataSize; idx += NeedleEntrySize {
+			line := resp.IndexFileContent[idx : idx+NeedleEntrySize]
+			key := BytesToNeedleId(line[:NeedleIdSize])
+			offset := BytesToOffset(line[NeedleIdSize : NeedleIdSize+OffsetSize])
+			size := util.BytesToUint32(line[NeedleIdSize+OffsetSize : NeedleIdSize+OffsetSize+SizeSize])
+			eachEntryFn(key, offset, size)
+		}
+
+		return nil
 	})
-	if err != nil {
-		return err
-	}
-	return nil
 }