Chapter9. gRPC develop - bindata client streaming

[robert-min]

임의의 데이터 송수신

• 임의의 데이터(.taz, .gz 등)을 처리하기 위해 bytes 타입을 사용해 처리

[robert-min]

임의의 데이터 송수신 Service

• message에 oneof를 사용하여 데이터 송수신하는 어느 한 시점에서 oneof로 지정한 그룹 중 하나의 필드 값만을 설정하도록 강제

syntax = "proto3";

option go_package = "github.com/handson-go/chap9/bindata_client_streaming/service";

service Repo {
    rpc CreateRepo (stream RepoCreateRequest) returns (RepoCreateReply) {}
}

message RepoContext {
    string create_id = 1;
    string name = 2;
}

message RepoCreateRequest {
    oneof body {
        RepoContext context = 1;
        bytes data = 2;
    }
}

message Repository {
    string id = 1;
    string name = 2;
    string url = 3;
}

message RepoCreateReply { 
    Repository repo = 1;
    int32 size = 2;
}

[robert-min]

임의의 데이터 송수신 Server

• oneof 키워드를 설정했기 때문에 RepoCreateRequest 객체의 경우 context 또는 data 필드 중 둘중 하나에만 값이 존재
• 어떤 필드에 값이 있는지 확인하기 위해 case 구문 사용

type repoService struct {
    svc.UnimplementedRepoServer
}

func (s *repoService) CreateRepo(stream svc.Repo_CreateRepoServer) error {
    var repoContext *svc.RepoContext
    var data []byte
    for {
        r, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            return status.Error(
                codes.Unknown,
                err.Error(),
            )
        }

        switch t := r.Body.(type) {
        case *svc.RepoCreateRequest_Context:
            repoContext = r.GetContext()
        case *svc.RepoCreateRequest_Data:
            b := r.GetData()
            data = append(data, b...)
        case nil:
            return status.Error(
                codes.InvalidArgument,
                "Message doesn't contain context or data",
            )
        default:
            return status.Errorf(
                codes.FailedPrecondition,
                "Unexpected message type: %s",
                t,
            )
        }
    }
    repo := svc.Repository{
        Name: repoContext.Name,
        Url: fmt.Sprintf(
            "https://example.com/%s/%s",
            repoContext.CreateId,
            repoContext.Name,
        ),
    }

    r := svc.RepoCreateReply{
        Repo: &repo,
        Size: int32(len(data)),
    }
    return stream.SendAndClose(&r)
}

func registerServices(s *grpc.Server) {
    svc.RegisterRepoServer(s, &repoService{})
}

func startServer(s *grpc.Server, l net.Listener) error {
    return s.Serve(l)
}

func main() {
    listenAddr := os.Getenv("LISTEN_ADDR")
    if len(listenAddr) == 0 {
        listenAddr = ":50051"
    }

    lis, err := net.Listen("tcp", listenAddr)
    if err != nil {
        log.Fatal(err)
    }
    s := grpc.NewServer()
    registerServices(s)
    log.Fatal(startServer(s, lis))
}

[robert-min]

임의의 데이터 송수신 Server test

func startTestGrpcServer() *bufconn.Listener {
    l := bufconn.Listen(1)
    s := grpc.NewServer()
    registerServices(s)
    go func() {
        startServer(s, l)
    }()
    return l
}

func TestCreateRepo(t *testing.T) {
    l := startTestGrpcServer()

    bufconnDialer := func(
        ctx context.Context, addr string,
    ) (net.Conn, error) {
        return l.Dial()
    }

    client, err := grpc.DialContext(
        context.Background(),
        "", grpc.WithInsecure(),
        grpc.WithContextDialer(bufconnDialer),
    )
    if err != nil {
        t.Fatal("DialContext", err)
    }

    // 1. 리포지터리 이름과 소유자 정보를 주고 받음
    repoClient := svc.NewRepoClient(client)
    stream, err := repoClient.CreateRepo(
        context.Background(),
    )
    if err != nil {
        t.Fatal("CreateRepo", err)
    }

    c := svc.RepoCreateRequest_Context{
        Context: &svc.RepoContext{
            CreateId: "user-123",
            Name:     "test-repo",
        },
    }
    r := svc.RepoCreateRequest{
        Body: &c,
    }
    err = stream.Send(&r)
    if err != nil {
        t.Fatal("StreamSend", err)
    }

    // 2. data 필드만 설정한 RepoCreateContext 객체를 전송
    data := "Arbitray data bytes"
    repoData := strings.NewReader(data)

    for {
        b, err := repoData.ReadByte()
        if err == io.EOF {
            break
        }

        bData := svc.RepoCreateRequest_Data{
            Data: []byte{b},
        }
        r := svc.RepoCreateRequest{
            Body: &bData,
        }
        err = stream.Send(&r)
        if err != nil {
            t.Fatal("StreaSend", err)
        }
        l.Close()
    }

    // 3. 서버에서 받아들인 응답이 정상인지 확인
    resp, err := stream.CloseAndRecv()
    if err != nil {
        t.Fatal("CloseAndRecv", err)
    }
    expectedSize := int32(len(data))
    if resp.Size != expectedSize {
        t.Errorf(
            "Expected Repo Created to be: %d bytes Got back: %d",
            expectedSize,
            resp.Size,
        )
    }
    expectedRepoUrl := "https://example.com/user-123/test-repo"
    if resp.Repo.Url != expectedRepoUrl {
        t.Errorf(
            "Expected Repo URL to be: %s, Got: %s",
            expectedRepoUrl,
            resp.Repo.Url,
        )
    }
}