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-ClientStreaming

1. brief introduction

above 【Golang | gRPC】HTTP Connection management —— from HTTP/1.0 To HTTP/2.0 Evolution of A brief introduction gRPC Mid stream mode is mainly divided into client stream 、 Server stream 、 Advantages of bidirectional streaming and streaming mode , Let's go through a demo To illustrate gRPC Use of client streams

2. practice

There is one scenario below ： The server stores the user's age information , Client pass stream Send a message containing the user's name multiple times message, Server through stream receive message, Return the age and... Of all requesting users at once

2.1 proto file

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

syntax = "proto3";
package pb;
option go_package= ".;pb";

//  Define the methods contained in the query service 
service Query {
  //  Client stream mode 
  rpc GetAge (stream userInfo) returns (ageInfo){}
}

//  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;
}

The server implements a query （Query） service , Contains a method GetAge. Method GetAge Add keywords before the input parameters of stream To indicate that this method enables client-side streaming

2.1.2 stay .\gRPC-ClientStreaming\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 grpc.pb.go file

2.2.1 see query_grpc.pb.go The interface definition of the client stream and the server stream generated in and the server QueryServer Definition of service

//  Client stream 
type Query_GetAgeClient interface {
    
	Send(*UserInfo) error
	CloseAndRecv() (*AgeInfo, error)
	grpc.ClientStream
}
//  Server stream 
type Query_GetAgeServer interface {
    
	SendAndClose(*AgeInfo) error
	Recv() (*UserInfo, error)
	grpc.ServerStream
}
// Query The client interface of the service 
type QueryClient interface {
    
	GetAge(ctx context.Context, opts ...grpc.CallOption) (Query_GetAgeClient, error)
}
// Query The server interface of the service 
type QueryServer interface {
    
	GetAge(Query_GetAgeServer) error
	mustEmbedUnimplementedQueryServer()
}

• Client stream use Send send out message, Use CloseAndRecv receive message
• client GetAge The first return value of the method is Query_GetAgeClient, Indicates that a stream is generated , For sending and receiving message; If there are multiple ways , Then each method can generate a flow
• Server side GetAge The input of the method is Query_GetAgeServer（ flow ）, The specific method needs to be implemented by the user , Can receive and send from the stream message

2.3 Server side

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

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

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

type Query struct {
    
	pb.UnimplementedQueryServer //  Version compatibility involved 
}

func (q *Query) GetAge(serverStream pb.Query_GetAgeServer) error {
    
	log.Println("start of stream")
	var names_received []*pb.UserInfo
	for {
    
		userinfoRecv, err := serverStream.Recv()
		//  After the client actively closes the stream , sign out for loop 
		if err == io.EOF {
    
			log.Println("end of the recv direction of the stream")
			break
		}
		log.Printf("The name of user received is %s\n", userinfoRecv.GetName())
		names_received = append(names_received, userinfoRecv)
	}
	//  Count age and 
	var ages_sum int32
	for _, v := range names_received {
    
		ages_sum += userinfo[v.GetName()]
	}
	//  return message
	log.Printf("send message about the total of ages:%d ", ages_sum)
	err := serverStream.SendAndClose(&pb.AgeInfo{
    Age: ages_sum})
	if err != nil {
    
		log.Panic(err)
	}
	log.Println("end of the send direction of the stream")
	return nil
}

• Every time the server receives one message, Save its user name , Until the client closes the stream in the sending direction
• Recv The method will block until stream Received in message, Or until the client calls CloseAndRecv Method
• When the client calls CloseAndRecv When the method is used , Server call Recv The method will get io.EOF Return value
• Server call SendAndClose Method to send a response message And close the flow in the sending direction

2.3.2 The service registers and starts

func main() {
    
	//  establish socket Monitor 
	listener, _ := net.Listen("tcp", ":1234")
	// new One gRPC The server , Used to register services 
	grpcserver := grpc.NewServer()
	//  Registration service method 
	pb.RegisterQueryServer(grpcserver, new(Query))
	//  Turn on gRPC service 
	_ = grpcserver.Serve(listener)
}

Use RegisterQueryServer This method is directed to gRPC The server Registered in the QueryServer

2.4 client

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

2.4.1 First establish a connection without authentication , Generate Client, And then through the method GetAge Return the corresponding stream , Finally, flow message Send and receive

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)
	// return GetAge Flow corresponding to method 
	queryStream, _ := client.GetAge(context.Background())
	//  towards stream Sent in message
	_ = queryStream.Send(&pb.UserInfo{
    Name: "foo"})
	time.Sleep(time.Second)
	_ = queryStream.Send(&pb.UserInfo{
    Name: "bar"})
	time.Sleep(time.Second)

	//  After sending data twice, actively close the stream and wait for receiving data from server Terminal message
	ages_sum, err := queryStream.CloseAndRecv()
	if err != nil {
    
		log.Println(err)
	}
	fmt.Printf("The total of ages of foo and bar is %d", ages_sum.GetAge())
}

• Client pass CloseAndRecv Method actively closes the stream in the sending direction while waiting to receive the message

The operation results are as follows ：

3. summary

• Client pass Send Method is sent multiple times message, adopt CloseAndRecv Method actively closes the stream in the sending direction while waiting to receive the message
• Server through Recv Method receive multiple times message, adopt SendAndClose Method to send a response message And close the flow in the sending direction