gRPC (3) - Server Streaming 개념 및 예제코드

Server Streaming

cf. 예제코드는 go 개발환경과 protoc-gen-go 디펜던시가 갖추어졌다는 전제로 작성되었다.

• Http2의 특징을 활용한, 새로운 형태의 RPC API.
• 클라이언트의 요청 한 번으로, 서버가 데이터를 여러 번 전송할 수 있는 형태.

 

---

 

실습환경은 아래와 같은 디렉토리에서 이루어졌다.

go를 설치하고 환경변수로 GOPATH를 등록한 상태로,

모든 명령어는 ~/go/src/ch4 에서 이루어졌다.

 

 

1. proto 파일 생성

greetpb 디렉토리에 greetpb.proto 파일을 생성하고 아래와 같이 입력해준다.

syntax = "proto3";

package greet;
option go_package="greet/greetpb";

message Greeting {
    string first_name = 1;
    string last_name = 2;
}

message GreetRequest {
    Greeting greeting = 1;
}

message GreetResponse {
    string result = 1;
}

message GreetManyTimesRequest {
    Greeting greeting = 1;
}

message GreetManyTimesResponse {
    string result = 1;
}

service GreetService {
    // Server Streaming API
    rpc GreetManyTimes(GreetManyTimesRequest) returns (stream GreetManyTimesResponse) {};
}

 

서버와 클라이언트에서 사용할 Request / Response 구조체 (Struct)를 message라는 이름으로 정의하고,

rpc 통신에 사용할 Service와 메소드를 정의한다. 메소드의 경우 파라미터는 input, returns 뒤에는 output을 넣어준다.

server streaming의 경우 서버에서 클라이언트로 응답을 여러 번 전송해야 하므로,

GreetManyTimes 메소드의 리턴값 앞에 stream이라는 설정을 붙여주는 것에 유의.

 

이후 터미널에

protoc greet/greetpb/greet.proto --go\_out=plugins=grpc:.

 

명령어를 입력하면 proto 파일을 토대로 go파일을 생성해준다 (greet.pb.go 파일이 생성된다)

 

---

2. server.go

package main

import (
	"ch4/greet/greetpb"
	"context"
	"fmt"
	"log"
	"net"

	"google.golang.org/grpc"
)

type server struct{}

func (*server) GreetManyTimes(
	req *greetpb.GreetManyTimesRequest,
	stream greetpb.GreetService_GreetManyTimesServer) error {
	firstName := req.GetGreeting().GetFirstName()
	for i := 0; i < 10; i++ {
		result := "Hello " + firstName + " number " + strconv.Itoa(i)
		res := &greetpb.GreetManyTimesResponse{
			Result: result,
		}
		stream.Send(res)
		time.Sleep(1000 * time.Millisecond)
	}
	return nil
}

func main() {
	fmt.Println("Hello world")

	// 50051: default port for gRPC. 1. tcp 연결을 위한 port binding
	lis, err := net.Listen("tcp", "0.0.0.0:50051")
	if err != nil {
		log.Fatalf("Failed to listen", err)
	}

	// 2. grpc server open
	s := grpc.NewServer()
	greetpb.RegisterGreetServiceServer(s, &server{})

	// port와 grpc server 연결 후 serving
	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to server: %v", err)
	}

}

 

proto에서 정의했던 GreetManyTimes 메소드를 구현한다.

편의상, 클라이언트가 firstName를 넣어서 서버에 요청을 보내면

서버에서 응답을 10번 반환하도록 했다.

---

3. Client.go

package main

import (
	"ch4/greet/greetpb"
	"context"
	"fmt"
	"io"
	"log"

	"google.golang.org/grpc"
)

func main() {
	fmt.Println("Hello, this is client")
	// 기본적으로는 ssl init을 지원하지만, ssl 작업 없이 시작하기 위해 insecure 옵션 사용
	// 1. create the connection
	conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
	if err != nil {
		log.Fatalf("could not connect %v", err)
	}

	// conn 커넥션이 프로그램 종료 시 같이 끊기도록 설정함. defer는 프로그램 말미에 실행되도록 하는 명령어
	// 3. close connection. (end of program)
	defer conn.Close()

	// 2. create a client
	c := greetpb.NewGreetServiceClient(conn)
	fmt.Printf("created client %f", c)

	doServerStreaming(c)
}

func doServerStreaming(c greetpb.GreetServiceClient) {
	fmt.Println("Starting to do a Server Streaming RPC.")

	req := &greetpb.GreetManyTimesRequest{
		Greeting: &greetpb.Greeting{
			FirstName: "firstName",
			LastName:  "LastName",
		},
	}
	resStream, err := c.GreetManyTimes(context.Background(), req) // it returns String, err 두 개의 값을 리턴함.
	if err != nil {
		log.Fatalln("Error while Calling GreetManyTimes ", err)
	}
	// stream이 닫히면 EOF 에러 리턴함.
	for {
		msg, err := resStream.Recv()
		if err == io.EOF {
			// reach the end of stream
			break
		}
		if err != nil {
			log.Fatalln("Err while reading stream. ", err)
		}
		log.Printf("Response from GreetManyTimes : ", msg.GetResult())
	}

}

 

서버가 stream 형태로 응답하므로, GreetManyTimes 메소드의 리턴값으로 resStream을 정의한 뒤

resStream의 Recv() 메소드로 응답을 받는다.

서버에서 전송을 마치면 EOF 에러를 반환하므로, EOF 에러를 받으면 stream을 종료한다.