go gRPC 服务端推送给客户端流demo

syntax = "proto3";

option go_package = "./;pb";

package pb;

service Greeter{
    rpc GetStream (StreamReqData) returns (stream StreamResData) {}
}

// 请求数据格式
message StreamReqData{
    string data = 1;
}

// 返回数据格式
message StreamResData{
    string data = 1;
}

client/main.go:

package main

import (
	"context"
	pb "demo01/pb"
	"log"
	"google.golang.org/grpc"
	_ "google.golang.org/grpc/balancer/grpclb"
)
const (
	ADDRESS = "localhost:50051"
)

func main(){
	// 建立连接
	conn, err := grpc.Dial(ADDRESS, grpc.WithInsecure())
	if err != nil {
		return
	}
	defer conn.Close()
	// 通过刚才的链接 生成一个client对象
	c := pb.NewGreeterClient(conn)

	// 通过服务端推送流
	reqstreamData := &pb.StreamReqData{Data:"aaa"}

	res, _ := c.GetStream(context.Background(),reqstreamData)
	for {
		aa, err := res.Recv()
		if err != nil {
			log.Println(err)
			return
		}
		log.Println(aa)
	}
}

server/main.go:

package main

import (
	pb "demo01/pb"
	"fmt"
	"net"
	"time"
	"google.golang.org/grpc"
)

const (
	port = ":50051"
)

// 服务对象
type server struct {
	pb.UnimplementedGreeterServer
}

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

}

func main(){
	lis, err := net.Listen("tcp", port)	
	if err != nil {
		return 
	}
	// 创建一个grpc 服务器
	s := grpc.NewServer()
	// 注册事件
	pb.RegisterGreeterServer(s, &server{})
	// 处理链接
	s.Serve(lis)
}