Akka-gRPC

Table of contents

Reading Time: 3 minutes

Akka gRPC provides support for building streaming gRPC servers and clients on top of Akka Streams and Akka Http.

Features of Akka-gRPC

A generator, that starts from a protobuf service definitions, for:
Model classes
The service API as a Scala trait using Akka Stream Sources
On the server side code to create an Akka HTTP route based on your implementation of the service
On the client side, a client for the service
gRPC Runtime implementation that uses
- Akka Http/2 support or the server side and
- grpc-netty-shaded for the client side.

Steps for generated Code

Define the protobuf file under src/main/protobuf folder . If we define file other than src main/protobuf folder the you we have to provide explicit path for .protobuf file under <proto> tag.
Compile the .proto file
Setup the server so that client send request to server
Provide the implementation for generated code on server side
Provide the implementation for generated code on client side
Start the server for maven project

mvn compile dependency:properties exec:exec@server

7. Start the client for maven project

mvn compile dependency:properties exec:exec@client

Define Protobuf file

syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.knoldus.helloworld";
option java_outer_classname = "HelloWorldKnoldus";

// The greeting service definition.
service GreeterService {
    // Sends a greeting
    rpc SayHello (HelloRequest) returns (HelloReply) {}
}
message HelloRequest {
    string name = 1;
}
message HelloReply {
    string message = 1;
}

After compiling .proto file , the code will be generated

Set-up Server

First, the GreeterServer main class creates an akka.actor.ActorSystem, a container in which Actors, Akka Streams and Akka HTTP run. Next, it defines a function from HttpRequest to Future[HttpResponse] using the GreeterServiceImpl. This function handles gRPC requests in the HTTP/2 with TLS server that is bound to port 8080 in this example.

import akka.actor.ActorSystem;
import akka.http.javadsl.*;
import akka.http.javadsl.model.HttpRequest;
import akka.http.javadsl.model.HttpResponse;
import akka.japi.Function;
import akka.stream.ActorMaterializer;
import akka.stream.Materializer;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.security.KeyFactory;
import java.security.KeyStore;
import java.security.PrivateKey;
import java.security.SecureRandom;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.security.spec.PKCS8EncodedKeySpec;
import java.util.Base64;
import java.util.concurrent.CompletionStage;


public class GreeterServer {

  public static void main(String[] args) throws Exception {
    // important to enable HTTP/2 in ActorSystem's config
    Config conf = ConfigFactory.parseString("akka.http.server.preview.enable-http2 = on")
      .withFallback(ConfigFactory.load());
    ActorSystem system = ActorSystem.create("HelloWorld", conf);
    new GreeterServer(system).run();
  }

  final ActorSystem system;

  public GreeterServer(ActorSystem system) {
    this.system = system;
  }

  public CompletionStage<ServerBinding> run() throws Exception {

    Materializer materializer = ActorMaterializer.create(system);

    Function<HttpRequest, CompletionStage<HttpResponse>> service =
        GreeterServiceHandlerFactory.create(
            new GreeterServiceImpl(materializer),
            system);

    CompletionStage<ServerBinding> bound =
        Http.get(system).bindAndHandleAsync(
          service,
          ConnectWithHttps.toHostHttps("127.0.0.1", 8080)
              .withCustomHttpsContext(serverHttpContext()),
          materializer
        );

    bound.thenAccept(binding ->
        System.out.println("gRPC server bound to: " + binding.localAddress())
    );

    return bound;
  }
}

Implementation on Server Side

For the server the following classes are generated:

Message classes, such as Hello Request and HelloReply
GreeterService interface of the service
GreeterServiceHandler utility to create the HttpRequest to HttpResponse function from the GreeterServiceImpl

The part that we have to implement on the server side is the GreeterServiceImpl which implements the generated GreeterService interface. It is this implementation that is bound to the HTTP server via the GreeterServiceHandler and it looks like this:

import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.Materializer;
import akka.stream.javadsl.BroadcastHub;
import akka.stream.javadsl.Keep;
import akka.stream.javadsl.MergeHub;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

class GreeterServiceImpl implements GreeterService {

  final Materializer materializer;
  final Sink<HelloRequest, NotUsed> inboundHub;
  final Source<HelloReply, NotUsed> outboundHub;

  public GreeterServiceImpl(Materializer materializer) {
    this.materializer = materializer;

  }

  @Override
  public CompletionStage<HelloReply> sayHello(HelloRequest request) {
    return CompletableFuture.completedFuture(
        HelloReply.newBuilder()
            .setMessage("Hello, " + request.getName())
            .build()
    );
  }

}

Implementation on client side

In this example we have the client in the same project as the server. That is common for testing purposes but for real usage you or another team would have a separate project (different service) that is using the client and doesn’t implement the server side of the service. Between such projects you would only share the proto file (by copying it).

From the same proto file that was used on the server side classes are generated for the client:

Message classes, such as HelloRequest and HelloReply
GreeterService interface of the service
GreeterServiceClient that implements the client side of the Greeter Service.

On the client side we don’t have to implement anything, the GreeterServiceClient is ready to be used as is.

We need an ActorSystem and then the GreeterServiceClient can bc created and used like this:

import akka.Done;
import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.Materializer;
import akka.actor.ActorSystem;
import akka.grpc.GrpcClientSettings;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.Source;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletionStage;

import static akka.NotUsed.notUsed;
class GreeterClient {

  public static void main(String[] args) {
    final ActorSystem sys = ActorSystem.create("HelloWorldClient");
    final Materializer materializer = ActorMaterializer.create(sys);

    GreeterServiceClient client = GreeterServiceClient.create(
        GrpcClientSettings.fromConfig("helloworld.GreeterService", sys),
        materializer,
        sys.dispatcher()
    );

    final List<String> names;
    if (args.length == 0) {
      names = Arrays.asList("Alice", "Bob");
    } else {
      names = Arrays.asList(args);
    }

    names.forEach(name -> {
      System.out.println("Performing request: " + name);
      HelloRequest request = HelloRequest.newBuilder()
          .setName(name)
          .build();
      CompletionStage<HelloReply> replyCS = client.sayHello(request);
      replyCS.whenComplete((reply, error) -> {
        if (error == null) {
          System.out.println(reply.getMessage());
        } else {
          System.out.println(error.getMessage());
        }
      });
    });
  }
}