Realization of four data flow modes of grpc based on Multilingual Communication

grpc There are many data transmission modes ：

• Simple mode ： The client initiates a request , The server responds to data once , And ordinary rpc There is no difference between .
• Server side data flow mode ： The client initiates a request , The server returns a continuous data stream . A typical example is that the client sends a stock code to the server , The server continuously returns the data of the stock to the client in real time . In addition, our commonly used subscription scenarios also belong to the service side stream mode .
• Client data stream mode . Contrary to the server-side data flow pattern , The client continuously sends data streams like the server , After sending , The server returns a response . A typical example is that the Internet of things terminal sends data to the server .
• Two way data flow pattern ： The client and server can send real-time data streams to both sides for interaction , Typical applications are chat robots , Instant messaging tools, etc .

propto file

Various modes of proto The contents of the document are as follows ：

syntax = "proto3";

option go_package = ".;proto";  //  Indicates the current directory 

service Greeter {
  rpc GetStream(StreamReqData)returns(stream StreamResData);  // Server stream mode 
  rpc PutStream(stream StreamReqData)returns(StreamResData);  // Client stream mode 
  rpc Al1Stream(stream StreamReqData)returns(stream StreamResData); // Two way flow mode 
}

message StreamReqData {
  string data = 1;
}
message StreamResData {
  string data = 1;
}

serve Server program

The server must implement proto Three functions in .

package main

import (
	"OldPackageTest/stream_grpc_test/proto"
	"fmt"
	"google.golang.org/grpc"
	"net"
	"sync"
	"time"
)

const PORT = ":50052"

type server struct {
}

//  Server stream mode 
func (s *server) GetStream(req *proto.StreamReqData, res proto.Greeter_GetStreamServer) error {
	i := 0
	for {
		i++
		_ = res.Send(&proto.StreamResData{
			Data: fmt.Sprintf("%v", time.Now().Unix()),
		})
		time.Sleep(time.Second)
		if i > 10 {
			break
		}
	}

	return nil
}



//  Client stream mode 
func (s *server) PutStream(cliStr proto.Greeter_PutStreamServer) error {
	for {
		if a, err := cliStr.Recv(); err != nil {
			fmt.Println(err)
			break
		} else {
			fmt.Println(a.Data)
		}
	}

	return nil
}


//  Two way flow mode 
func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error {
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		defer wg.Done()
		for {
			data, _ := allStr.Recv()
			fmt.Println(" Received client message ：" + data.Data)
		}
	}()

	go func() {
		defer wg.Done()
		for {
			_ = allStr.Send(&proto.StreamResData{Data: " I'm the server "})
			time.Sleep(time.Second)
		}
	}()

	wg.Wait()
	return nil
}

func main() {
	lis, err := net.Listen("tcp", PORT)
	if err != nil {
		panic(err)
	}
	s := grpc.NewServer()
	proto.RegisterGreeterServer(s, &server{})
	err = s.Serve(lis)
	if err != nil {
		panic(err)
	}
}

client Client program

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"google.golang.org/grpc"

	"OldPackageTest/stream_grpc_test/proto"
)

func main() {
	conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())
	if err != nil {
		panic(err)
	}
	defer conn.Close()
	// Server stream mode 
	c := proto.NewGreeterClient(conn)
	res, _ := c.GetStream(context.Background(), &proto.StreamReqData{Data: " For class network "})
	for {
		a, err := res.Recv() // If you know socket Programming words will understand  send recv
		if err != nil {
			fmt.Println(err)
			break
		}
		fmt.Println(a.Data)
	}

	// Client stream mode 
	putS, _ := c.PutStream(context.Background())
	i := 0
	for {
		i++
		_ = putS.Send(&proto.StreamReqData{
			Data: fmt.Sprintf(" For class network %d", i),
		})
		time.Sleep(time.Second)
		if i > 10 {
			break
		}
	}

	// Two way flow mode 
	allStr, _ := c.AllStream(context.Background())
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		defer wg.Done()
		for {
			data, _ := allStr.Recv()
			fmt.Println(" Received client message ：" + data.Data)
		}
	}()

	//1.  Concentrated learning protobuf, grpc

	go func() {
		defer wg.Done()
		for {
			_ = allStr.Send(&proto.StreamReqData{Data: " For class network "})
			time.Sleep(time.Second)
		}
	}()

	wg.Wait()
}