syntax = "proto3";
option go_package = "./;proto";
service Greeter {
    // 定義方法，stream是常量，流模式
    rpc ServerStream (StreamRequestData) returns (stream StreamResponseData);      //服務端流模式，拉消息
    rpc ClientStream (stream StreamRequestData) returns (StreamResponseData);      //客戶端流模式，推消息
    rpc AllStream (stream StreamRequestData) returns (stream StreamResponseData);  //雙向流模式，能推能拉
}
message StreamRequestData {
    string data = 1; //編號
}
message StreamResponseData {
    string data = 1; //編號
}

package main
import (
	"file_test/grpc_go_stream/proto"
	"fmt"
	"net"
	"sync"
	"time"
	"google.golang.org/grpc"
)
const port = 8082
type server struct{}
func (s *server) ServerStream(req *proto.StreamRequestData, res proto.Greeter_ServerStreamServer) error {
	i := 0
	for {
		i++
		//業(yè)務代碼
		_ = res.Send(&proto.StreamResponseData{
			Data: fmt.Sprintf("這是發(fā)給%s的數(shù)據流", req.Data),
		})
		time.Sleep(time.Second * 1)
		if i > 10 {
			break
		}
	}
	return nil
}
func (s *server) ClientStream(cliStr proto.Greeter_ClientStreamServer) error {
	for {
		//業(yè)務代碼
		res, err := cliStr.Recv()
		if err != nil {
			fmt.Println("本次客戶端流數(shù)據發(fā)送完了:",err)
			break
		}
		fmt.Println("客戶端發(fā)來消息：",res.Data)
	}
	return nil
}
func (s *server) AllStream(allStr proto.Greeter_AllStreamServer) error {
	wg:=sync.WaitGroup{}
	wg.Add(2)
	//接受客戶端消息的協(xié)程
	go func() {
		defer wg.Done()
		for  {
			//業(yè)務代碼
			res, err := allStr.Recv()
			if err != nil {
				fmt.Println("本次客戶端流數(shù)據發(fā)送完了:",err)
				break
			}
			fmt.Println("收到客戶端發(fā)來消息：",res.Data)
		}
	}()
	//發(fā)送消息給客戶端的協(xié)程
	go func() {
		defer wg.Done()
		i := 0
		for {
			i++
			//業(yè)務代碼
			_ = allStr.Send(&proto.StreamResponseData{
				Data: fmt.Sprintf("這是發(fā)給客戶端的數(shù)據流"),
			})
			time.Sleep(time.Second * 1)
			if i > 10 {
				break
			}
		}
	}()
	wg.Wait()
	return nil
}
// 啟動
func start() {
	// 1.實例化server
	g := grpc.NewServer()
	// 2.注冊邏輯到server中
	proto.RegisterGreeterServer(g, &server{})
	// 3.啟動server
	lis, err := net.Listen("tcp", "127.0.0.1:8082")
	if err != nil {
		panic("監(jiān)聽錯誤:" + err.Error())
	}
	err = g.Serve(lis)
	if err != nil {
		panic("啟動錯誤:" + err.Error())
	}
}
func main() {
	start()
}

package main
import (
	"context"
	"file_test/grpc_go_stream/proto"
	"fmt"
	"sync"
	"time"

	"google.golang.org/grpc"
)
var rpc proto.GreeterClient
func serverStreamDemo()  {
	//服務端流模式
	res,err:=rpc.ServerStream(context.Background(),&proto.StreamRequestData{Data: "jeff"})
	if err != nil {
		panic("rpc請求錯誤："+err.Error())
	}
	for  {
		data,err:=res.Recv() //
		if err != nil {
			fmt.Println("客戶端發(fā)送完了:",err)
			return
		}
		fmt.Println("客戶端返回數(shù)據流值:",data.Data)
	}
}
func clientStreamDemo()  {
	//客戶端流模式
	cliStr, err := rpc.ClientStream(context.Background())
	if err != nil {
		panic("rpc請求錯誤：" + err.Error())
	}
	i := 0
	for {
		i++
		_ = cliStr.Send(&proto.StreamRequestData{
			Data: "jeff",
		})
		time.Sleep(time.Second * 1)
		if i > 10 {
			break
		}
	}
}
func clientAndServerStreamDemo()  {
	//雙向流模式
	allStr, _ := rpc.AllStream(context.Background())
	wg := sync.WaitGroup{}
	wg.Add(1)
	//接受服務端消息的協(xié)程
	go func() {
		defer wg.Done()
		for {
			//業(yè)務代碼
			res, err := allStr.Recv()
			if err != nil {
				fmt.Println("本次服務端流數(shù)據發(fā)送完了:", err)
				break
			}
			fmt.Println("收到服務端發(fā)來消息：", res.Data)
		}
	}()
	//發(fā)送消息給服務端的協(xié)程
	go func() {
		defer wg.Done()
		i := 0
		for {
			i++
			//業(yè)務代碼
			_ = allStr.Send(&proto.StreamRequestData{
				Data: fmt.Sprintf("這是發(fā)給服務端的數(shù)據流"),
			})
			time.Sleep(time.Second * 1)
			if i > 10 {
				break
			}
		}
	}()
	wg.Wait()
}
// 啟動
func start() {
	conn, err := grpc.Dial("127.0.0.1:8082", grpc.WithInsecure())
	if err != nil {
		panic("rpc連接錯誤：" + err.Error())
	}
	defer conn.Close()
	rpc = proto.NewGreeterClient(conn) //初始化
	serverStreamDemo() //服務端流模式
	clientStreamDemo()  //客戶端流模式
	clientAndServerStreamDemo() // 雙向流模式
}
func main() {
	start()
}