Go gRPC 튜토리얼 #2 - Server Streaming RPC

이번 글에서는 Go gRPC의 Server Stream RPC에 대해 다룹니다

이전 글 보기
Go gRPC 튜토리얼 #1 - 세팅부터 Unary RPC Call 사용까지

 

Server Streaming RPC

클라이언트가 서버에 리퀘스트를 보내면 서버는 스트림을 보내게 된다.
리턴되는 스트림이 없을 때 까지 읽는다.

또한 gRPC는 개별적인 RPC call에 대해 메세지 순서를 보증한다.

이제 예제를 통해 Server Stream RPC을 알아보자
이번 글에선 Unary RPC인 GetInfo 함수와 Server Stream RPC인 ListInfo 함수를 만들어 볼 것이다.

 

먼저 프로토버퍼를 정의한다

ProtoBuffer 정의

syntax = "proto3";

package v1;
option go_package = "proto/v1";

service Route {
  rpc GetInfo(Content) returns (Content) {}
  // Unary RPC

  rpc ListInfo(Content) returns (stream Content) {}
  // Server Stream RPC
}

message Content {
  string message = 1;
}
// Content 타입에 대한 정의

그리고 아래 커맨드대로 protoc을 이용해 *.pb.go 형태의 go파일로 생성한다
protoc -I ./ info.proto --go_out=plugins=grpc:.

Server Stream RPC 서버 구현

우선 gRPC 핸들러를 제외한 서버를 작성해본다.

package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "io/ioutil"
    "log"
    "net"

    pb "aaronroh.com/m/proto/v1"
    "google.golang.org/grpc"
)

var (
    port     = flag.Int("port", 10000, "The server port")
    jsonFile = flag.String("json_file", "", "Json file containing list of content")
)
// flag 를 통해 option을 지정할 수 있다.
// flag 정의에 대한 내용은 go run server.go --help 를 통해 볼 수 있다.

type RouteServer struct {
    pb.UnimplementedRouteServer
  savedContents []*pb.Content
  // 전송할 내용에 대해 담을 어레이 공간이다.
  // loadContents 함수를 통해 json_file 의 내용이 저장된다
}

func main() {
    flag.Parse()

    lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", *port))

    if err != nil {
        log.Fatalf("Failed to listen %v", err)
    }

    s := &RouteServer{}
    s.loadContents(*jsonFile)

    grpcServer := grpc.NewServer()
    pb.RegisterRouteServer(grpcServer, s)

    if err := grpcServer.Serve(lis); err != nil {
        log.Fatalf("%v", err)
    }
}

func (s *RouteServer) loadContents(filePath string) {
    if filePath == "" {
        log.Fatalf("Must set jsonFile option")
    }

    data, err := ioutil.ReadFile(filePath)

    if err != nil {
        log.Fatalf("Failed to load Contents: %v", err)
    }

    if err := json.Unmarshal(data, &s.savedContents); err != nil {
        log.Fatalf("Failed to load: %v", err)
    }
}

 

이제 RPC 핸들러를 작성한다

func (s *RouteServer) GetInfo(ctx context.Context, req *pb.Content) (*pb.Content, error) {
    log.Printf("GetInfo - %v", req)
    return &pb.Content{Message: "Hi!"}, nil
}
// Unary RPC

func (s *RouteServer) ListInfo(req *pb.Content, stream pb.Route_ListInfoServer) error {
    log.Printf("ListInfo - %v", req)

    for _, content := range s.savedContents {
    // loadContents를 통해 저장된 데이터를 사용
        if err := stream.Send(content); err != nil {
      // Unary와 달리 stream.Send 를 이용해 클라이언트에 데이터를 보낸다.
            return err
        }
    }
    return nil
}
// Server Stream RPC

 

서버 단의 코드는 모두 완성되었다.
옵션으로 건네 줄 Json File은 아래 내용으로 data.json 이라고 저장하자

[
  {
  "Message": "Hi"
  },
  {
    "Message": "My name is ~"
  }
]

go run server.go --json_file data.json 으로 서버를 실행할 수 있다.
실행을 해도 아직 클라이언트의 요청이 없기에 로그는 아직 뜨지 않는다

클라이언트 구현

package main

import (
    "context"
    "flag"
    "io"
    "log"

    pb "aaronroh.com/m/proto/v1"

    "google.golang.org/grpc"
)

var serverAddr = flag.String("server_addr", "localhost:10000", "The server address with port")

func main() {
    flag.Parse()

  conn, err := grpc.Dial(*serverAddr, grpc.WithInsecure())
  // 예제용 이기에 별도로 인증서를 사용하지 않았다. 따라서 WithInsecure() 로 설정

    if err != nil {
        log.Fatalf("failed to dial: %v", err)
    }

    defer conn.Close()

    client := pb.NewRouteClient(conn)

  content, err := client.GetInfo(context.Background(), &pb.Content{Message: "Hi GetInfo Unary RPC"})
  // Unary RPC인 GetInfo에 대한 요청

    if err != nil {
        log.Fatalf("%v", err)
    }

    log.Printf("%s", content)

  stream, err := client.ListInfo(context.Background(), &pb.Content{Message: "Hi ListInfo Server Stream RPC"})
  // Server Stream RPC인 ListInfo에 대한 요청

    if err != nil {
        log.Fatalf("ListInfo - %v", err)
    }

    for {
    content, err := stream.Recv()
    // Stream은 Unary와 달리 Recv() 를 이용해 값을 받는다
    // for문을 통해 하나씩 풀어본다.
        if err == io.EOF {
            break
    }
    // Stream 데이터를 모두 받았는지 확인을 위해 io.EOF 를 이용한다

        if err != nil {
            log.Fatalf("ListInfo stream - %v", err)
        }

    log.Printf("Content: Message: %s", content.GetMessage())
    }
}

클라이언트 단의 코드도 모두 작성하였다
go run client.go를 통해 실행할 수 있다.

 

이제 서버를 가동한 채로 클라이언트에서 요청을 보내보자

JSON 파일의 내용대로 스트림 형태의 전송이 이루어짐을 확인할 수 있다.

(좌측: 서버 로그, 우측: 클라이언트 로그)

 

이번 내용의 예제는 go-grpc-server_streaming-example 에서 확인할 수 있다.