gRPC (4) - Client Streaming 개념 및 예제코드

Client Streaming

 

 

 

클라이언트가 여러 번 서버에 데이터 전송, 클라이언트가 응답을 언제 받을 수 있는지는 전적으로 서버에게 달려있다.

 

---

 

 

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

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 LongGreetRequest {
    Greeting greeting = 1;
}

message LongGreetResponse {
    string result = 1;
}

service GreetService {
    // Client Streaming API
    rpc LongGreet(stream LongGreetRequest) returns (LongGreetResponse) {};
}

클라이언트에서 데이터를 여러 번 보내야 하므로, LongGreet 메소드의 파라미터로 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"
	"io"
	"log"
	"net"
	"strconv"
	"time"

	"google.golang.org/grpc"
)

type server struct{}

func (*server) LongGreet(stream greetpb.GreetService_LongGreetServer) error {
	fmt.Println("LongGreet function was invoked with stream")
	result := "Hello "
	for {
		req, err := stream.Recv()
		if err == io.EOF {
			// 전송 끝. return result
			// 만약 sendAndClose가 제대로 작동하지 않는다 -> err 리턴. 따라서 그대로 return해도 된다.
			return stream.SendAndClose(&greetpb.LongGreetResponse{
				Result: result,
			})
		}
		if err != nil {
			log.Fatalf("Error while reading client stream", err)
		}
		firstName := req.GetGreeting().GetFirstName()
		result += firstName + " "
	}
}

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에서 정의했던 LongGreet 메소드를 구현한다.

클라이언트가 여러 번 요청을 보내면, 서버가 요청을 취합해서 한 번 값을 리턴하는 구조로 되어 있다.

 

---

3. Client.go

package main

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

	"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)

	doClientStreaming(c)
}


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

	requests := []*greetpb.LongGreetRequest{
		&greetpb.LongGreetRequest{
			Greeting: &greetpb.Greeting{
				FirstName: "firstName 1",
			},
		},
		&greetpb.LongGreetRequest{
			Greeting: &greetpb.Greeting{
				FirstName: "firstName 2",
			},
		},
		&greetpb.LongGreetRequest{
			Greeting: &greetpb.Greeting{
				FirstName: "firstName 3",
			},
		},
		&greetpb.LongGreetRequest{
			Greeting: &greetpb.Greeting{
				FirstName: "firstName 4",
			},
		},
	}

	// streaming의 특성상, request가 메소드 호출에 직접 들어가지는 않는다.
	stream, err := c.LongGreet(context.Background())
	if err != nil {
		log.Fatalf("Error while calling stream ", err)
	}

	// iterate over our slice, send each msg individually
	for _, req := range requests {
		fmt.Println("Sending request ", req)
		stream.Send(req)
		time.Sleep(1000 * time.Millisecond)
	}

	// 마지막 데이터를 보내고, 서버에게서 '제대로 받았다'는 응답이 오기까지 대기.
	res, err := stream.CloseAndRecv()
	if err != nil {
		log.Fatalf("Error while receiving response ", err)
	}
	fmt.Println("LongGreet Response : ", res)
}

클라이언트가 streaming 형태로 데이터를 전송해야 하므로, LongGreet 메소드의 리턴값으로 stream을 정의한 뒤

for문으로 stream.Send() 를 사용해서 데이터를 전송한다.

마지막 데이터를 보낸 뒤에는 서버에게 CloseAndRecv() 메소드를 보내서

데이터 전송이 끝났음 + 서버의 응답을 기다린다.