gRPC Learning (Part 1)

Basics, Unary and Server Streaming

Overview

Use ProtoBuf to define the contract(*.proto) between services and message(request/response). The contract interface will be generated and to be implemented by the developer.

ProtoBuf Advantages:

1. Language agnostic, code can be generated for any target languages;
2. Data is in binary format and can be efficiently serialized(payload is small)
3. Convenient for large data transportation
4. Allows easy API evolution using rules (?)
5. gRPC is the future of micro-service api and mobile-server api

gRPC Basics

gRPC is used to define the Message(request and response); and Service(endpoint and service name) Sample *.proto contract as below:

gRPC server is async by default, no blocking threads on request, each server can serve millions of requests in parallel gRPC client can be sync or async, flexible in either mode and can perform client side load balancing (?).

Comparison of gRPC and JSON

1. Small Size. Reduce CPU consumption for JSON parsing
2. ProtoBuf(binary) is more efficient and faster, good for mobile platform.
3. ProtoBuf contract is generated automatically on the fly.
4. ProtoBuf messages allows communicating other micro services in different languages, there’s no languages restrictions in each micro services.

HTTP/2

HTTP2 is much faster than conventional HTTP

HTTP1.1 opens new TCP connection to server at each request, does not compress header, only works on req/res pattern, no server push is allowed.
Headers in plain text, each req/res exchange involves many overhead in transmission.

HTTP2 allows client and server push messages in parallel in same TCP connection, reduces latency. Support server push and header compression , it’s via binary protocol, faster and efficient , It’s also secure.

In HTTP2, once the single TCP connection is established, server and client starts multiple roundtrip for subsequent communications in same TCP connection.

Types of gRPC APIs

1. Unary
2. Server Streaming
3. Client Streaming
4. Bi-directional String

gRPC and REST

gRPC hands-on

Setting up gradle dependencies

Note:

1. Add plugins(com.google.protobuf and idea);
2. Define protobuf plugin and idea plugin to include the generated source classpath.

group 'com.yangli907.learning.grpc'
version '1.0-SNAPSHOT'

apply plugin: 'java'
apply plugin: 'com.google.protobuf'
apply plugin: 'idea'

sourceCompatibility = 1.8

buildscript {
    repositories {
        mavenCentral()
    }
    dependencies {
        // ASSUMES GRADLE 2.12 OR HIGHER. Use plugin version 0.7.5 with earlier
        // gradle versions
        classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.10'
    }
}


repositories {
    mavenCentral()
}

dependencies {

    // grpc
    compile 'io.grpc:grpc-netty-shaded:1.12.0' // shaded: includes ssl libraries
    compile 'io.grpc:grpc-protobuf:1.12.0'
    compile 'io.grpc:grpc-stub:1.12.0'
    compile "io.grpc:grpc-services:1.12.0"   // reflection

    testCompile group: 'junit', name: 'junit', version: '4.12'

    // https://mvnrepository.com/artifact/org.mongodb/mongodb-driver-sync
    compile group: 'org.mongodb', name: 'mongodb-driver-sync', version: '3.8.2'

}


protobuf {
    protoc {
        artifact = "com.google.protobuf:protoc:3.5.1-1"
    }
    plugins {
        grpc {
            artifact = 'io.grpc:protoc-gen-grpc-java:1.12.0'
        }
    }
    generateProtoTasks {
        all()*.plugins {
            grpc {}
        }
    }
}

// if you have source imports issues, add the below
sourceSets.main.java.srcDir new File(buildDir, 'generated/source')
idea {
    module {
        // Marks the already(!) added srcDir as "generated"
        generatedSourceDirs += file('build/generated/source')
    }
}

Sample *.proto file for simple greeting application can be defined as below, and it’ll generate sources upon “generate protobuf” task execution

syntax = "proto3";
package dummy;
option java_package = "com.proto.dummy";
option java_multiple_files = true;
message DummyMessage {}
service DummyService {}

gRPC unary definiton

ProtoBuf schema is defined as below:

syntax = "proto3";

package dummy;

option java_package = "com.proto.greet";

option java_multiple_files = true;

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

message GreetRequest {
    Greeting greeting = 1;
}

message GreetResponse {
    string result = 1;
}

service GreetService {
    // Unary
    rpc Greet(GreetRequest) returns (GreetResponse) {};
}

service implementation needs to extend the auto-generated *ServiceImplBase class, which has the method interface defined as specified in *.proto file’s “service” section.

Streaming API

Client sends one message, server returns a stream of messages back. Using stream keyword in thee proto file definition . The service definition follows the same pattern as unary mode, only difference is the response can be returned many times in the service method implementation.

For example, below is the application to return a stream of messages upon receive of one client request:

@Override
public void greetManyTimes(GreetManyTimesRequest request, StreamObserver responseObserver) {
    String firstName = request.getGreeting().getFirstName();

    try {
        for (int i = 0; i < 10; i++) {
            String result = "Hello " + firstName + ", response number: " + i;
            GreetManyTimesResponse response = GreetManyTimesResponse.newBuilder().setResult(result).build();

            responseObserver.onNext(response);
            Thread.sleep(1000);
        }
    }
     catch (InterruptedException e) {
            e.printStackTrace();
        }
    finally {
        responseObserver.onCompleted();
    }
}

• [ ] TODO: Investigate whether a client logic change would require server side restart? → Per experiment locally, it seems the server needs to be restarted when client implementation is changed.

Exercise: Decompose Prime Number

Full implementation below: Server

public class DecomposePrimeServer {

    public static void main(String args[]) throws IOException, InterruptedException {
        System.out.println("hello gRPC");
        Server server = ServerBuilder.forPort(50051)
                                     .addService(new DecomposePrimeServiceImpl())
                                     .build();

        server.start();

        Runtime.getRuntime().addShutdownHook(new Thread(
            () -> {
                System.out.println("Received shutdown request.");
                server.shutdown();
                System.out.println("Server stopped.");

            }
        ));

        server.awaitTermination();
    }
}

Client

public class DecomposePrimeClient {

    public static void main(String args[]) {
        System.out.println("Hello gRPC Client");

        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 50051)
                                                      .usePlaintext()
                                                      .build();

        System.out.println("Creating Stub");

        DecomposePrimeGrpc.DecomposePrimeBlockingStub syncPrimeClient = DecomposePrimeGrpc.newBlockingStub(channel);

        PrimeRequest primeRequest = PrimeRequest.newBuilder().setNumber(32).build();

        syncPrimeClient.decomposePrime(primeRequest).forEachRemaining(
            r -> System.out.println("value is: " + r.getNumber())
        );

        System.out.println("Shutting down channel");
        channel.shutdown();
    }
}

ServiceImpl

public class DecomposePrimeServiceImpl extends DecomposePrimeGrpc.DecomposePrimeImplBase {
    @Override
    public void decomposePrime(PrimeRequest request, StreamObserver responseObserver) {

        int input = request.getNumber();
        int divisor = 2;

        try {
            while(input > 1) {
                    if (input % divisor == 0) {
                        PrimeResponse response = PrimeResponse.newBuilder().setNumber(divisor).build();

                        responseObserver.onNext(response);

                        input /= divisor;
                    } else {
                        divisor += 1;
                    }

                Thread.sleep(1000);

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            responseObserver.onCompleted();
        }
    }
}