Implement the communication models of gRPC - - Alibaba Cloud Documentation Center

Sample project

For information about the sample project of gRPC, see hello-servicemesh-grpc. The directories in this topic are directories of hello-servicemesh-grpc.

Step 1: Convert code

Run the following command to install gRPC and Protocol Buffers. In this example, gRPC and Protocol Buffers are installed in the macOS operating system.
```
brew install grpc protobuf
```
Covert the Protocol Buffers definition to code in the programming languages that you use. In the topic, Java, Go, Node.js, and Python are used:
Note In the sample project, the code directory of each language contains the proto directory that stores the landing.proto file. The landing.proto file is a symbolic link to the proto/landing.proto file in the root directory of the sample project. This way, you can update the Protocol Buffers definition in a unified manner.
- Java: Maven is a build automation tool for Java. Maven provides the protobuf-maven-plugin plug-in to automatically convert code. You can run the mvn package command to use protoc-gen-grpc-java to generate gRPC template code. For more information, see hello-grpc-java/pom.xml.
- Go: Run the go get github.com/golang/protobuf/protoc-gen-go command to install protoc-gen-go. Then, run the protoc command to generate gRPC code. For more information, see hello-grpc-go/proto2go.sh.
- Node.js: Run the npm install -g grpc-tools command to install grpc_tools_node_protoc. Then, run the protoc command to generate gRPC code. For more information, see hello-grpc-nodejs/proto2js.sh.
- Python: Run the pip install grpcio-tools command to install grpcio-tools. Then, run the protoc command to generate gRPC code. For more information, see hello-grpc-python/proto2py.sh.

Step 2: Set communication models

Set the hello array.

Java:

private final List<String> HELLO_LIST = Arrays.asList("Hello", "Bonjour", "Hola", "こんにちは", "Ciao", "안녕하세요");
kv.put("data", HELLO_LIST.get(index));

Go:

var helloList = []string{"Hello", "Bonjour", "Hola", "こんにちは", "Ciao", "안녕하세요"}
kv["data"] = helloList[index]

Node.js:

let hellos = ["Hello", "Bonjour", "Hola", "こんにちは", "Ciao", "안녕하세요"]
kv.set("data", hellos[index])

Python:

hellos = ["Hello", "Bonjour", "Hola", "こんにちは", "Ciao", "안녕하세요"]
result.kv["data"] = hellos[index]

Set the communication models.

Set the unary RPC model.

Java:

// Use the blocking stub to send a request to the server.
public TalkResponse talk(TalkRequest talkRequest) {
    return blockingStub.talk(talkRequest);
}
// After the server processes the request, the onNext and onCompleted events of the StreamObserver instance are triggered.
public void talk(TalkRequest request, StreamObserver<TalkResponse> responseObserver) {
    ...
    responseObserver.onNext(response);
    responseObserver.onCompleted();
}

Go:

func talk(client pb.LandingServiceClient, request *pb.TalkRequest) {
    r, err := client.Talk(context.Background(), request)
}

func (s *ProtoServer) Talk(ctx context.Context, request *pb.TalkRequest) (*pb.TalkResponse, error) {
    return &pb.TalkResponse{
        Status:  200,
        Results: []*pb.TalkResult{s.buildResult(request.Data)},
    }, nil
}

Node.js:

function talk(client, request) {
    client.talk(request, function (err, response) {
            ...
    })
}
                
function talk(call, callback) {
    const talkResult = buildResult(call.request.getData())
    ...
    callback(null, response)
}

Python:

def talk(stub):
    response = stub.talk(request)
    
def talk(self, request, context):
    result = build_result(request.data)
    ...
    return response

Set the server streaming RPC model.

Java:

public List<TalkResponse> talkOneAnswerMore(TalkRequest request) {
    Iterator<TalkResponse> talkResponses = blockingStub.talkOneAnswerMore(request);
    talkResponses.forEachRemaining(talkResponseList::add);
    return talkResponseList;
}

public void talkOneAnswerMore(TalkRequest request, StreamObserver<TalkResponse> responseObserver) {
    String[] datas = request.getData().split(",");
    for (String data : datas) {...}
    talkResponses.forEach(responseObserver::onNext);
    responseObserver.onCompleted();
}

Go:

func talkOneAnswerMore(client pb.LandingServiceClient, request *pb.TalkRequest) {
    stream, err := client.TalkOneAnswerMore(context.Background(), request)
    for {
        r, err := stream.Recv()
        if err == io.EOF {
            break
        }
    ...
    }
}

func (s *ProtoServer) TalkOneAnswerMore(request *pb.TalkRequest, stream pb.Landing..Server) error {
    datas := strings.Split(request.Data, ",")
    for _, d := range datas {
        stream.Send(&pb.TalkResponse{...})
}

Node.js:

function talkOneAnswerMore(client, request) {
    let call = client.talkOneAnswerMore(request)
    call.on('data', function (response) {
        ...
    })
}

function talkOneAnswerMore(call) {
    let datas = call.request.getData().split(",")
    for (const data in datas) {
        ...
        call.write(response)
    }
    call.end()
}

Python:

def talk_one_answer_more(stub):
    responses = stub.talkOneAnswerMore(request)
    for response in responses:
        logger.info(response)

def talkOneAnswerMore(self, request, context):
    datas = request.data.split(",")
    for data in datas:
        yield response

Set the client streaming RPC model.

Java:

public void talkMoreAnswerOne(List<TalkRequest> requests) throws InterruptedException {
    final CountDownLatch finishLatch = new CountDownLatch(1);
    StreamObserver<TalkResponse> responseObserver = new StreamObserver<TalkResponse>() {
        @Override
        public void onNext(TalkResponse talkResponse) {
            log.info("Response=\n{}", talkResponse);
        }
        @Override
        public void onCompleted() {
            finishLatch.countDown();
        }
    };
    final StreamObserver<TalkRequest> requestObserver = asyncStub.talkMoreAnswerOne(responseObserver);
    try {
        requests.forEach(request -> {
            if (finishLatch.getCount() > 0) {
                requestObserver.onNext(request);
        });
    requestObserver.onCompleted();
}
                         
public StreamObserver<TalkRequest> talkMoreAnswerOne(StreamObserver<TalkResponse> responseObserver) {
    return new StreamObserver<TalkRequest>() {
        @Override
        public void onNext(TalkRequest request) {
            talkRequests.add(request);
        }
        @Override
        public void onCompleted() {
            responseObserver.onNext(buildResponse(talkRequests));
            responseObserver.onCompleted();
        }
    };
}

Go:

func talkMoreAnswerOne(client pb.LandingServiceClient, requests []*pb.TalkRequest) {
    stream, err := client.TalkMoreAnswerOne(context.Background())
    for _, request := range requests {
        stream.Send(request)
    }
    r, err := stream.CloseAndRecv()
}

func (s *ProtoServer) TalkMoreAnswerOne(stream pb.LandingService_TalkMoreAnswerOneServer) error {
    for {
        in, err := stream.Recv()
        if err == io.EOF {
            talkResponse := &pb.TalkResponse{
                Status:  200,
                Results: rs,
            }
            stream.SendAndClose(talkResponse)
            return nil
        }
        rs = append(rs, s.buildResult(in.Data))
    }
}

Node.js:

function talkMoreAnswerOne(client, requests) {
    let call = client.talkMoreAnswerOne(function (err, response) {
        ...
    })
    requests.forEach(request => {
        call.write(request)
    })
    call.end()
}

function talkMoreAnswerOne(call, callback) {
    let talkResults = []
    call.on('data', function (request) {
        talkResults.push(buildResult(request.getData()))
    })
    call.on('end', function () {
        let response = new messages.TalkResponse()
        response.setStatus(200)
        response.setResultsList(talkResults)
        callback(null, response)
    })
}

Python:

def talk_more_answer_one(stub):
    response_summary = stub.talkMoreAnswerOne(request_iterator)

def generate_request():
    for _ in range(0, 3):
        yield request
        
def talkMoreAnswerOne(self, request_iterator, context):
    for request in request_iterator:
        response.results.append(build_result(request.data))
    return response

Set the bidirectional streaming RPC model.

Java:

public void talkBidirectional(List<TalkRequest> requests) throws InterruptedException {
    final CountDownLatch finishLatch = new CountDownLatch(1);
    StreamObserver<TalkResponse> responseObserver = new StreamObserver<TalkResponse>() {
        @Override
        public void onNext(TalkResponse talkResponse) {
            log.info("Response=\n{}", talkResponse);
        }
        @Override
        public void onCompleted() {
            finishLatch.countDown();
        }
    };
    final StreamObserver<TalkRequest> requestObserver = asyncStub.talkBidirectional(responseObserver);
    try {
        requests.forEach(request -> {
            if (finishLatch.getCount() > 0) {
                requestObserver.onNext(request);
    ...
    requestObserver.onCompleted();
}

public StreamObserver<TalkRequest> talkBidirectional(StreamObserver<TalkResponse> responseObserver) {
    return new StreamObserver<TalkRequest>() {
        @Override
        public void onNext(TalkRequest request) {
            responseObserver.onNext(TalkResponse.newBuilder()
                    .setStatus(200)
                    .addResults(buildResult(request.getData())).build());
        }
        @Override
        public void onCompleted() {
            responseObserver.onCompleted();
        }
    };
}

Go:

func talkBidirectional(client pb.LandingServiceClient, requests []*pb.TalkRequest) {
    stream, err := client.TalkBidirectional(context.Background())
    waitc := make(chan struct{})
    go func() {
        for {
            r, err := stream.Recv()
            if err == io.EOF {
                // read done.
                close(waitc)
                return
            }
        }
    }()
    for _, request := range requests {
        stream.Send(request)
    }
    stream.CloseSend()
    <-waitc
}

func (s *ProtoServer) TalkBidirectional(stream pb.LandingService_TalkBidirectionalServer) error {
    for {
        in, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        stream.Send(talkResponse)
    }
}

Node.js:

function talkBidirectional(client, requests) {
    let call = client.talkBidirectional()
    call.on('data', function (response) {
        ...
    })
    requests.forEach(request => {
        call.write(request)
    })
    call.end()
}

function talkBidirectional(call) {
    call.on('data', function (request) {
        call.write(response)
    })
    call.on('end', function () {
        call.end()
    })
}

Python:

def talk_bidirectional(stub):
    responses = stub.talkBidirectional(request_iterator)
    for response in responses:
        logger.info(response)

def talkBidirectional(self, request_iterator, context):
    for request in request_iterator:
        yield response

Step 3: Implement functions

Implement the environment variable function.

Java:

private static String getGrcServer() {
    String server = System.getenv("GRPC_SERVER");
    if (server == null) {
        return "localhost";
    }
    return server;
}

Go:

func grpcServer() string {
    server := os.Getenv("GRPC_SERVER")
    if len(server) == 0 {
        return "localhost"
    } else {
        return server
    }
}

Node.js:

function grpcServer() {
    let server = process.env.GRPC_SERVER;
    if (typeof server !== 'undefined' && server !== null) {
        return server
    } else {
        return "localhost"
    }
}

Python:

def grpc_server():
    server = os.getenv("GRPC_SERVER")
    if server:
        return server
    else:
        return "localhost"

Implement the random number function.

Java:

public static String getRandomId() {
    return String.valueOf(random.nextInt(5));
}

Go:

func randomId(max int) string {
    return strconv.Itoa(rand.Intn(max))
}

Node.js:

function randomId(max) {
    return Math.floor(Math.random() * Math.floor(max)).toString()
}

Python:

def random_id(end):
    return str(random.randint(0, end))

Implement the timestamp function.

Java:

TalkResult.newBuilder().setId(System.nanoTime())

Go:
```
result.Id = time.Now().UnixNano()
```

Node.js:

result.setId(Math.round(Date.now() / 1000))

Python:
```
result.id = int((time.time()))
```

Implement the UUID function.

Java:

kv.put("id", UUID.randomUUID().toString());

Go:

import (
    "github.com/google/uuid"
)
kv["id"] = uuid.New().String()

Node.js:
```
kv.set("id", uuid.v1())
```
Python:
```
result.kv["id"] = str(uuid.uuid1())
```

Implement the sleep function.

Java:
```
TimeUnit.SECONDS.sleep(1);
```
Go:
```
time.Sleep(2 * time.Millisecond)
```

Node.js:

let sleep = require('sleep')
sleep.msleep(2)

Python:
```
time.sleep(random.uniform(0.5, 1.5))
```

Verify the results

Feature

Run the following commands to start the gRPC server on a terminal and the gRPC client on another terminal. After you start the gRPC client and server, the gRPC client sends requests to the API operations of the four communication models.

Java:

mvn exec:java -Dexec.mainClass="org.feuyeux.grpc.server.ProtoServer"

mvn exec:java -Dexec.mainClass="org.feuyeux.grpc.client.ProtoClient"

Go:

go run server.go

go run client/proto_client.go

Node.js:

node proto_server.js

node proto_client.js

Python:

python server/protoServer.py

python client/protoClient.py

If no communication error occurs, the gRPC client and server are started.

Cross communication

Cross communication ensures that the gRPC client and server communicate with each other in the same manner, no matter what language is used by the gRPC client and server. This way, the response of a request does not vary with the language version.

Start the gRPC server, for example, the Java gRPC server:

mvn exec:java -Dexec.mainClass="org.feuyeux.grpc.server.ProtoServer"

Run the following commands to start the gRPC clients in Java, Go, Node.js, and Python:
```
mvn exec:java -Dexec.mainClass="org.feuyeux.grpc.client.ProtoClient"
```
```
go run client/proto_client.go
```
```
node proto_client.js
```
```
python client/protoClient.py
```
If no communication error occurs, cross communication is successful.

What to do next

After you verify that the gRPC client and server can communicate as expected, you can build images for the client and server.

Step 1: Build a project

Use four programming languages to build projects for the gRPC client and server.

Java

Create JAR packages for the gRPC client and server. Then, copy the packages to the Docker directory.

mvn clean install -DskipTests -f server_pom
cp target/hello-grpc-java.jar ../docker/

mvn clean install -DskipTests -f client_pom
cp target/hello-grpc-java.jar ../docker/

Go
The binary files that are compiled by using Go contain the configuration about the operating systems and need to be deployed in Linux. Therefore, add the following content to the binary file: Then, copy the binary files to the Docker directories.
```
env GOOS=linux GOARCH=amd64 go build -o proto_server server.go
mv proto_server ../docker/

env GOOS=linux GOARCH=amd64 go build -o proto_client client/proto_client.go
mv proto_client ../docker/
```
NodeJS
The Node.js project must be created in a Docker image to support all kinds of C++ dependencies that are required for the runtime. Therefore, copy the file to the Docker directory.
```
cp ../hello-grpc-nodejs/proto_server.js node
cp ../hello-grpc-nodejs/package.json node
cp -R ../hello-grpc-nodejs/common node
cp -R ../proto node
cp ../hello-grpc-nodejs/*_client.js node
```

Python

Copy the Python file to the Docker directory without compilation.

cp -R ../hello-grpc-python/server py
cp ../hello-grpc-python/start_server.sh py
cp -R ../proto py
cp ../hello-grpc-python/proto2py.sh py
cp -R ../hello-grpc-python/client py
cp ../hello-grpc-python/start_client.sh py

Step 2: Build images for the gRPC server and client

After you build the project, all the files that are required by Dockerfile are saved in the Docker directory. This section describes the major information about the Dockerfile.

Select alpine as the basic image because its size is the smallest. In the example, the basic image of Python is python v2.7. You can change the image version as needed.
Node.js requires the installation of C++ and the compiler Make. The Npm package needs to be installed with grpc-tools.

This example shows how to build the image of the Node.js server.

Create the grpc-server-node.dockerfile file.

FROM node:14.11-alpine
RUN apk add --update \
      python \
      make \
      g++ \
  && rm -rf /var/cache/apk/*
RUN npm config set registry http://registry.npmmirror.com && npm install -g node-pre-gyp grpc-tools --unsafe-perm
COPY node/package.json .
RUN npm install --unsafe-perm
COPY node .
ENTRYPOINT ["node","proto_server.js"]

Build an image.

docker build -f grpc-server-node.dockerfile -t registry.cn-beijing.aliyuncs.com/asm_repo/grpc_server_node:1.0.0 .

A total of eight images are built.

Run the Push command to distribute the images to Container Registry.

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_server_java:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_client_java:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_server_go:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_client_go:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_server_node:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_client_node:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_server_python:1.0.0

docker push registry.cn-beijing.aliyuncs.com/asm_repo/grpc_client_python:1.0.0