refactor grpc API

weed/pb/master.proto

@@ -9,7 +9,7 @@ option go_package = "github.com/chrislusf/seaweedfs/weed/pb/master_pb";
 service Seaweed {
   rpc SendHeartbeat (stream Heartbeat) returns (stream HeartbeatResponse) {
   }
-    rpc KeepConnected (stream KeepConnectedRequest) returns (stream VolumeLocation) {
+  rpc KeepConnected (stream KeepConnectedRequest) returns (stream KeepConnectedResponse) {
   }
   rpc LookupVolume (LookupVolumeRequest) returns (LookupVolumeResponse) {
   }
@@ -142,6 +142,18 @@ message VolumeLocation {
   uint32 grpc_port = 7;
 }
 
+message ClusterNodeUpdate {
+  string node_type = 1;
+  string address = 2;
+  bool is_leader = 3;
+  bool is_add = 4;
+}
+
+message KeepConnectedResponse {
+  VolumeLocation volume_location = 1;
+  ClusterNodeUpdate cluster_node_update = 2;
+}
+
 message LookupVolumeRequest {
   repeated string volume_or_file_ids = 1;
   string collection = 2; // optional, a bit faster if provided.

weed/server/master_grpc_server.go

@@ -203,7 +203,7 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 	}()
 
 	for _, message := range ms.Topo.ToVolumeLocations() {
-		if sendErr := stream.Send(message); sendErr != nil {
+		if sendErr := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); sendErr != nil {
 			return sendErr
 		}
 	}
@@ -223,7 +223,7 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ
 	for {
 		select {
 		case message := <-messageChan:
-			if err := stream.Send(message); err != nil {
+			if err := stream.Send(&master_pb.KeepConnectedResponse{VolumeLocation: message}); err != nil {
 				glog.V(0).Infof("=> client %v: %+v", clientName, message)
 				return err
 			}
@@ -244,8 +244,10 @@ func (ms *MasterServer) informNewLeader(stream master_pb.Seaweed_KeepConnectedSe
 		glog.Errorf("topo leader: %v", err)
 		return raft.NotLeaderError
 	}
-	if err := stream.Send(&master_pb.VolumeLocation{
+	if err := stream.Send(&master_pb.KeepConnectedResponse{
+		VolumeLocation: &master_pb.VolumeLocation{
 			Leader: string(leader),
+		},
 	}); err != nil {
 		return err
 	}

weed/wdclient/masterclient.go

@@ -118,36 +118,39 @@ func (mc *MasterClient) tryConnectToMaster(master pb.ServerAddress) (nextHintedL
 		mc.currentMaster = master
 
 		for {
-			volumeLocation, err := stream.Recv()
+			resp, err := stream.Recv()
 			if err != nil {
 				glog.V(0).Infof("%s masterClient failed to receive from %s: %v", mc.clientType, master, err)
 				return err
 			}
 
+			if resp.VolumeLocation != nil {
 				// maybe the leader is changed
-			if volumeLocation.Leader != "" {
-				glog.V(0).Infof("redirected to leader %v", volumeLocation.Leader)
-				nextHintedLeader = pb.ServerAddress(volumeLocation.Leader)
+				if resp.VolumeLocation.Leader != "" {
+					glog.V(0).Infof("redirected to leader %v", resp.VolumeLocation.Leader)
+					nextHintedLeader = pb.ServerAddress(resp.VolumeLocation.Leader)
 					return nil
 				}
 
 				// process new volume location
 				loc := Location{
-				Url:        volumeLocation.Url,
-				PublicUrl:  volumeLocation.PublicUrl,
-				DataCenter: volumeLocation.DataCenter,
-				GrpcPort:   int(volumeLocation.GrpcPort),
+					Url:        resp.VolumeLocation.Url,
+					PublicUrl:  resp.VolumeLocation.PublicUrl,
+					DataCenter: resp.VolumeLocation.DataCenter,
+					GrpcPort:   int(resp.VolumeLocation.GrpcPort),
 				}
-			for _, newVid := range volumeLocation.NewVids {
+				for _, newVid := range resp.VolumeLocation.NewVids {
 					glog.V(1).Infof("%s: %s masterClient adds volume %d", mc.clientType, loc.Url, newVid)
 					mc.addLocation(newVid, loc)
 				}
-			for _, deletedVid := range volumeLocation.DeletedVids {
+				for _, deletedVid := range resp.VolumeLocation.DeletedVids {
 					glog.V(1).Infof("%s: %s masterClient removes volume %d", mc.clientType, loc.Url, deletedVid)
 					mc.deleteLocation(deletedVid, loc)
 				}
 			}
 
+		}
+
 	})
 	if gprcErr != nil {
 		glog.V(1).Infof("%s masterClient failed to connect with master %v: %v", mc.clientType, master, gprcErr)