Environmental Science ：
Golang： go1.18.2 windows/amd64
gRPC： v1.47.0
Protobuf： v1.28.0

Complete code ：
https://github.com/WanshanTian/GolangLearning
cd GolangLearning/RPC/gRPC-Watch

1. brief introduction

The client can use the Watch Mechanism to subscribe to the data or state of a node on the server , When it changes, you can receive the corresponding notice .
I studied a while ago gRPC Service end stream mode , Today we will use this flow pattern to realize Watch function

2. practice

There is one scenario below ： The server stores the user's age information , The client sends message You can get the age of the corresponding user or update the age of the corresponding user （ Age +1）; adopt Watch The function can monitor the user's age status in real time , When the age of a user changes , The client receives notification

2.1 proto file

2.1.1 newly build gRPC-Watch Folder , Use go mod init initialization , establish pb Folder , newly build query.proto file

syntax = "proto3";
package pb;
option go_package= ".;pb";
import "google/protobuf/empty.proto";

//  Define the methods contained in the query service 
service Query {
  rpc GetAge (userInfo) returns (ageInfo) {};
  rpc Update (userInfo) returns (google.protobuf.Empty) {};
  rpc Watch (watchTime) returns (stream userInfo){}
}

//  Request structure , Contains a name Field 
message userInfo {
  string name = 1;
}

//  Respond to the structure of the application , Contains a age Field 
message ageInfo {
  int32 age = 1;
}

// watch Time for 
message watchTime{
  int32 time = 1;
}

• GetAge and Update Methods are used to obtain age and update age respectively , Simple RPC The way
• Watch Method is used to monitor changes in age status , Adopt the service end flow mode
• When gRPC The method of does not require a request message Or no response is required message when , You can start with import "google/protobuf/empty.proto", Then use it directly google.protobuf.Empty

2.1.2 stay .\gRPC-Watch\pb Use... In the directory protoc Tools to compile , stay pb Generate directly under the folder .pb.go and _grpc.pb.go file . About protoc You can view the detailed use of 【Golang | gRPC】 Use protoc compile .proto file

protoc --go_out=./ --go-grpc_out=./ .\query.proto

2.2 Server side

stay gRPC-Watch New under the directory Server Folder , newly build main.go file

2.2.1 So let's go through Query This structure is specifically implemented QueryServer Interface

var userinfo = map[string]int32{
    
	"foo": 18,
	"bar": 20,
}

// Query  Structure , Realization QueryServer Interface 
type Query struct {
    
	mu                          sync.Mutex
	ch                          chan string
	pb.UnimplementedQueryServer //  Version compatibility involved 
}

func (q *Query) GetAge(ctx context.Context, info *pb.UserInfo) (*pb.AgeInfo, error) {
    
	age := userinfo[info.GetName()]
	var res = new(pb.AgeInfo)
	res.Age = age
	return res, nil
}
//Update Used to update user age , adopt sync.Mutex Lock , If the age is updated , to chan Send the corresponding user name 
func (q *Query) Update(ctx context.Context, info *pb.UserInfo) (*emptypb.Empty, error) {
    
	q.mu.Lock()
	defer q.mu.Unlock()
	name := info.GetName()
	userinfo[name] += 1
	if q.ch != nil {
    
		q.ch <- name
	}
	return &emptypb.Empty{
    }, nil
}
//Watch Used to monitor the change of user's age status , Instantiate a chan, And then through select Method monitor chan Whether there is data in ,
// If yes, send it to the client through the server message, If there is no update after the specified time , The exit 
func (q *Query) Watch(timeSpecify *pb.WatchTime, serverStream pb.Query_WatchServer) error {
    
	if q.ch != nil {
    
		return errors.New("Watching is running, please stop first")
	}
	q.ch = make(chan string, 1)
	for {
    
		select {
    
		case <-time.After(time.Second * time.Duration(timeSpecify.GetTime())):
			close(q.ch)
			q.ch = nil
			return nil
		case nameModify := <-q.ch:
			log.Printf("The name of %s is updated\n", nameModify)
			serverStream.Send(&pb.UserInfo{
    Name: nameModify})
		}
	}
}

• Update Used to update user age , adopt sync.Mutex Lock , To prevent conflict ; If the age is updated and watch Function of open , to chan Send the corresponding user name
• Watch Used to monitor the change of user's age status , Instantiate a chan, Open for indication watch function . And then through select Method monitor chan Whether there is data in , If yes, send it to the client through the server message, If the age is not updated beyond the specified time , Then close watch Function and exit
• When Watch When the function has been turned on , If it is opened again, an error will be returned

2.3 client

stay gRPC-Watch New under the directory Client Folder , newly build main.go file

func main() {
    
	// Establish a connection without authentication 
	conn, err := grpc.Dial(":1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
    
		log.Panic(err)
	}
	defer conn.Close()
	client := pb.NewQueryClient(conn)
	//RPC Method call 
	ctx := context.Background()
	// Get the age before the update first 
	age, _ := client.GetAge(ctx, &pb.UserInfo{
    Name: "foo"})
	log.Printf("Before updating, the age is %d\n", age.GetAge())
	// Update age 
	log.Println("updating")
	client.Update(ctx, &pb.UserInfo{
    Name: "foo"})
	// Get the updated age 
	age, _ = client.GetAge(ctx, &pb.UserInfo{
    Name: "foo"})
	log.Printf("After updating, the age is %d\n", age.GetAge())
}

2.4 Watch function

stay gRPC-Watch New under the directory Watch Folder , newly build main.go file

func main() {
    
	// Establish a connection without authentication 
	conn, err := grpc.Dial(":1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
    
		log.Panic(err)
	}
	defer conn.Close()
	client := pb.NewQueryClient(conn)
	//RPC Method call 
	stream, _ := client.Watch(context.Background(), &pb.WatchTime{
    Time: 10})
	for {
    
		userInfoRecv, err := stream.Recv()
		if err == io.EOF {
    
			log.Println("end of watch")
			break
		} else if err != nil {
    
			log.Println(err)
			break
		}
		log.Printf("The name of %s is updated\n", userInfoRecv.GetName())
	}
}

2.5 function

Start the server first , Then open Watch function , When opening the client , There are the following output results （ When there is no age update within the specified time ,Watch Automatically quit ）：

When Watch Already open , And turn it on again , The following customized error messages will be returned , Corresponding 2.2 In code return errors.New("Watching is running, please stop first")

3. summary

• Watch The overall idea of the function is ： The server instantiates a chan, If the listening object changes , towards chan Send value in , The client from chan Received value in , If not, it will keep blocking