使用gRPC创建简单聊天程序

import (
	pb "grpc_chat/grpc_chat"
	"context"
	"log"
)

var MyNames = &pb.Names{}

var MyFriends = []*pb.Names{}

type chatService struct{}

func (cs *chatService) ExchangeNames(ctx context.Context, names *pb.Names) (*pb.Names, error){
	MyFriends = append(MyFriends, names);
	log.Printf("%s %s is comming for exchange names\n", names.FirstName, names.SecondName)
	return MyNames, nil;
}

import (
	pb "grpc_chat/grpc_chat"
	"context"
	"log"
	"google.golang.org/grpc/reflection"
	"net"
	"google.golang.org/grpc"
)

var Listen = ":8080"

func startService(){
	lis, err := net.Listen("tcp", ":17000")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer()
	pb.RegisterChatServiceServer(s, &chatService{})
	reflection.Register(s)
	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

func init() {
	flag.StringVar(&MyNames.FirstName, "fn", "", "first name")
	flag.StringVar(&MyNames.SecondName, "sn", "", "second name")
	flag.StringVar(&Listen, "l", ":8080", "listen address")
	flag.Parse()
}

func main() {
	go startService();
	select {};
}

type ChatServiceServer interface {
	ExchangeNames(context.Context, *Names) (*Names, error)
	GetList(*Count, ChatService_GetListServer) error
	SendList(ChatService_SendListServer) error
	Chat(ChatService_ChatServer) error
}

func (cs *chatService)GetList(c *pb.Count, gls pb.ChatService_GetListServer)error{
	// todo
	return nil
}
func (cs *chatService)SendList(sls pb.ChatService_SendListServer)error {
	// todo
	return nil
}

func (cs *chatService)Chat(scs pb.ChatService_ChatServer) error{
	// todo
	return nil
}

package main

import (
	"google.golang.org/grpc"
	"flag"
	pb "grpc_chat/grpc_chat"
	"log"
	"time"
	"context"
)
var (
	Server string
	MyNames = &pb.Names{}
	client pb.ChatServiceClient
)

func init(){
	flag.StringVar(&Server, "server", "127.0.0.1:8080", "grpc server address")
	flag.StringVar(&MyNames.FirstName, "fn", "", "first name")
	flag.StringVar(&MyNames.SecondName, "sn", "", "second name")
	flag.Parse()

	if conn, err := grpc.Dial(Server, grpc.WithInsecure()); err != nil {
		log.Fatal("connect to server failed", err)
	}else{
		client = pb.NewChatServiceClient(conn)
	}
}

func main(){
	ctx, cancel := context.WithTimeout(context.Background(), 1000*time.Second)
	defer cancel()
	sNames, err := client.ExchangeNames(ctx, MyNames);
	if err != nil {
		log.Fatalf("%v.ExchangeNames(_) = _, %v", client, err)
	}
	log.Printf("greet ok, server name: %s %s\n", sNames.FirstName, sNames.SecondName)
}

func (cs *chatService) GetList(c *pb.Count, gls pb.ChatService_GetListServer) error {
	MyFriendsLock.RLock();
	defer MyFriendsLock.RUnlock();
	if c.Offset > int32(len(MyFriends)) {
		return errors.New("no more friends")
	}
	count := c.Offset+c.Count;
	if count > int32(len(MyFriends)){
		count = int32(len(MyFriends)
	}
	sendNames := MyFriends[c.Offset:count]
	for _, names := range sendNames {
		if err := gls.Send(names); err != nil{
			return err;
		}
	}
	return nil
}

// 实际无意义，用来接收客户端发上来的列表
func (cs *chatService) SendList(sls pb.ChatService_SendListServer) error {
	var ReciveNames []*pb.Names;
	for {
		names, err := sls.Recv()
		if err == io.EOF {
			return sls.SendAndClose(&pb.Count{
				int32(len(ReciveNames)),
				0,
			})
		}
		if err != nil {
			return err
		}
		ReciveNames = append(ReciveNames, names)
		log.Println("send :", names.FirstName, names.SecondName)
	}
}

// 用于记录有多少客户端在chat
type Clients struct {
	sync.RWMutex
	Msg map[pb.ChatService_ChatServer]chan *pb.Msg
}
var clients = &Clients{Msg: map[pb.ChatService_ChatServer]chan *pb.Msg{}}

func (c *Clients) add(scs pb.ChatService_ChatServer, sendMsg chan *pb.Msg) {
	clients.Lock()
	clients.Msg[scs] = sendMsg
	clients.Unlock()
}

func (c *Clients) del(scs pb.ChatService_ChatServer) {
	clients.Lock()
	delete(clients.Msg, scs)
	clients.Unlock()
}

// 聊天双向流rpc
func (cs *chatService) Chat(scs pb.ChatService_ChatServer) error {
	sendMsg := make(chan *pb.Msg, 10)
	clients.add(scs, sendMsg)
	defer func() {
		clients.del(scs)
	}()
	waitc := make(chan struct{})
	go func() {
		for {
			in, err := scs.Recv()
			if err == io.EOF {
				close(waitc)
				return
			}
			if err != nil {
				log.Printf("receive error : %v\n", err)
				close(waitc)
				return
			}
			log.Printf(">>>>%s:%s\n", in.GetSender(), in.GetMsg())
			broadMsg(in)
		}
	}()
loop2:
	for {
		select {
		case s := <-sendMsg:
			scs.Send(s)
		case <-waitc:
			break loop2
		}
	}
	return nil
}

var sendMsg = make(chan string, 10);

func chat() {
	stream, err := client.Chat(context.Background())
	if err != nil {
		log.Fatalf("%v.Chat(_) = _, %v", client, err)
	}
	waitc := make(chan struct{})
	// 读协程
	go func() {
		for {
			in, err := stream.Recv()
			if err == io.EOF {
				close(waitc)
				return
			}
			if err != nil {
				log.Fatalf("Failed to receive a msg : %v", err)
			}
			log.Printf(">>>>%s:%s", in.GetSender(), in.GetMsg())
		}
	}()
	// 写协程
loop2:
	for {
		select {
		case s := <-sendMsg:
			if s == "exit" {
				stream.CloseSend();
				return;
			}
			stream.Send(&pb.Msg{
				MyNames.FirstName + "." + MyNames.SecondName,
				s,
			})
		case <-waitc:
			break loop2
		}
	}
}

func get() {
	stream, err := client.GetList(context.Background(), &pb.Count{10, 0});
	if err != nil {
		log.Fatalf("getlist error :%v", err)
	}
	for {
		in, err := stream.Recv()
		if err == io.EOF {
			return
		}
		if err != nil {
			log.Printf("Failed to receive a list : %v", err)
			return
		}
		log.Printf("==names===%s:%s", in.FirstName, in.SecondName)
	}
}
func send() {
	stream, err := client.SendList(context.Background())
	if err != nil {
		log.Fatalf("send list error :%v", err)
	}
	for i := 0; i < 10; i++ {
		err := stream.Send(&pb.Names{strconv.Itoa(i), strconv.Itoa(i)})
		if err != nil {
			log.Print("send error:", err)
			return
		}
	}
	c, err := stream.CloseAndRecv()
	if err != nil {
		log.Println("closeAndRecv failed:", err)
	} else {
		log.Println("send success", c.Count, c.Offset);
	}
}
func input() {
	for {
		var cmd, s string
		fmt.Scan(&cmd, &s)
		fmt.Println(cmd, s);
		switch cmd {
		case "get":
			get()
		case "send":
			send()
		case "msg":
			sendMsg <- s;
		case "exit":
			os.Exit(0);
		}
	}
}

package main

import (
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc"
	"log"
	pb "grpc_chat/grpc_chat"

	"net"
	"google.golang.org/grpc/grpclog"
	"context"
	"fmt"
	"time"
)

type tlsService struct{}

func (ts *tlsService) SendMsg(ctx context.Context, msg *pb.Msg) (*pb.Msg, error) {
	log.Printf("%s >>> %s\n", msg.Sender, msg.Msg);
	return &pb.Msg{"server", "receive"}, nil
}

func server() {
	listen, err := net.Listen("tcp", ":8080")
	if err != nil {
		grpclog.Fatalf("failed to listen: %v", err)
	}

	// TLS认证
	creds, err := credentials.NewServerTLSFromFile("./server.pem", "./server.key")
	if err != nil {
		grpclog.Fatalf("Failed to generate credentials %v", err)
	}

	// 实例化grpc Server, 并开启TLS认证
	s := grpc.NewServer(grpc.Creds(creds))

	// 注册HelloService
	pb.RegisterTlsServiceServer(s, &tlsService{})

	go s.Serve(listen)
}

func client() {
	creds, err := credentials.NewClientTLSFromFile("./server.pem", "a.com");
	if err != nil {
		log.Fatalln("NewClientTLSFromFile error", err)
	}
	conn, _ := grpc.Dial("localhost:8080", grpc.WithTransportCredentials(creds))
	// error handling omitted
	client := pb.NewTlsServiceClient(conn)
	msg, err := client.SendMsg(context.Background(), &pb.Msg{"client", "hello server"})
	fmt.Println(msg, err)
}

func main() {
	server();
	time.Sleep(time.Second)
	client();
}

	ctx := metadata.NewOutgoingContext(context.Background(), metadata.Pairs("auth", "pass1234"))
	msg, err := client.SendMsg(ctx, &pb.Msg{"client", "hello server"})

func unaryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	md, ok := metadata.FromIncomingContext(ctx);
	if !ok {
		return nil, errors.New("unaryInterceptor FromIncomingContext not ok ")
	}else if pass, ok := md["pass"]; !ok || pass[0] != "pass1234" {
		return nil, errors.New("auth failed");
	}
	return handler(ctx, req)
}

	s := grpc.NewServer(grpc.Creds(creds), grpc.UnaryInterceptor(unaryInterceptor))