gRPC Client-Implementierung in Java

Für Client-Aufrufe zu einem Service hat man 3 Möglichkeiten:

Ein Stub benötigt einen Channel zum Verbinden:

ManagedChannel channel = ManagedChannelBuilder
	.forAddress("localhost", 1234)
	.usePlaintext() // nur für Testzwecke
	.build();
gRPC Server-Implementierung in Java
public class GrpcServer {
	public static void main(String[] args) {
 
		Server server = ServerBuilder.forPort(9876)
			// Hier können mehrere Services hinzugefügt werden
			.addService(new FooServiceImpl())
			.build();
 
		try {
			server.start();
			System.out.println("Server running...");
			server.awaitTermination(); // blockierende Methode
		} catch (IOException | InterruptedException e) {
			// Fehlerbehebung
			e.printStackTrace();
		}
	}
}
Protobuf-Services in Java

Protoc generiert für jeden im protobuf-file definierte Service eine Klasse mit Grpc-Endung, die eine ImplBase-Klasse bereit stellt. Diese kann erweitert werden und darin die Service-Logik implementiert werden.

Anstatt die Antwort per return zurückzugeben, erwartet die Funktion zusätzlich zu den Parametern einen StreamObserver<T>, der die Methoden onNext(T), onError(Throwable) und onCompleted() bereitstellt.

Beispiel

public class FooServiceImpl extends FooServiceGrpc.FooServiceImplBase {
	@Override
	public void registerUser(Credentials c, StreamObserver<User> observer) {
		// Service-Logik...
		
		User user = User.newBuilder().build();
		
		observer.onNext(response);
		observer.onComplete();
	}
}

Siehe auch

Protobuf-Objekte in Java

Siehe auch

Erstellung

Neue Objektinstanzen werden mit dem Builder-Pattern erstellt:

T object = T.newBuilder()
	.setX(...)
	.setY(...)
	.build();

Beispiel

User user = User.newBuilder()
	.setUsername("john")
	.setAge(23)
	.addInterests(Interest.WORKING)
	.setAddress(Adress.newBuilder()
		.setStreet("Galgenbergstraße")
		.setPlz(93051)
		.build()
	)
	.build();

Attribute lesen

Attributwerte lassen sich mit gettern lesen:

user.getUsername();
Protobuf Servicedefinition

Siehe auch 

service FooService {
	rpc registerUser (User) returns (Nothing)
}

Streaming

Sowohl Parameter als auch Rückgabewerte können gestreamt werden. Dafür wird vor den Typen das Schlüsselwort stream geschrieben:

rpc ding (stream Foo) returns (stream Bar);
gRPC Nicht-blockierender Stub in Java

Siehe auch 

FooServiceGrpc.FooServiceStub stub = FooServiceGrpc.newStub(channel);
 
Foo foo = Foo.newBuilder().build();
 
stub.ding(foo, new BarObserver());

Der Bar-Observer muss implementiert werden: gRPC StreamObserver in Java

gRPC StreamObserver in Java
class Observer implements StreamObserver<Bar> {
	@Override
	public void onNext(Foo foo) {
		System.out.println("Received: " + foo.toString());
	}
 
	@Override
	public void onCompleted() {
		System.out.println("RPC call completed");
	}
 
	@Override
	public void onError(Throwable throwable) {
		System.out.println("RPC call error: " + throwable.getMessage());
	}
}
gRPC Stub mit Future in Java

Siehe auch 

FooServiceGrpc.FooServiceFutureStub futureStub =
	FooServiceGrpc.newFutureStub(channel);
 
Foo foo = Foo.newBuilder().build();
ListenableFuture<Bar> future = futureStub.ding(foo);
 
future.addListener(
	new Runnable() {
		@Override
		public void run() {
			// es gab eine Aktion des Servers
			System.out.println("done? " + future.isDone());
			System.out.println("canceled? " + future.isCancelled());
			if(future.isDone()) {
				try {
					Bar result = future.get();
				} catch (InterruptedException e) {
					// Fehlerbehandlung
				} catch (ExecutionException e) {
					// Fehlerbehandlung, siehe spätere Folie(n)
				}
			}
		}
	},
	executor
);
gRPC Blockierender Stub in Java

Siehe auch

ding blockiert so lange, bis der Rückgabetyp zurückgegeben wird.

FooServiceGrpc.FooServiceBlockingStub blockingStub =
	FooServiceGrpc.newBlockingStub(channel);
	
Foo foo = Foo.newBuilder().build();
 
Bar response = blockingStub.ding(foo);

Siehe auch

gRPC Streaming in Java

Streaming des Rückgabewertes

rpc ding (Foo) returns (stream Bar);

Konsequenzen:

  • Services
    • Bleiben syntaktisch gleich
    • onNext kann aber jetzt beliebig oft aufgerufen werden
  • Clients
    • Können nur noch asynchron sein

Streaming der Funktionsparameter

rpc dong (stream Foo) returns (Bar);

Konsequenzen:

  • Services

    • Service-Funktionen liefern jetzt einen StreamObserver zurück:

      public StreamObserver<Foo> dong(StreamObserver<Bar> responseObserver)
  • Clients
    • Können nur noch asynchron sein
    • Der Aufruf des Stubs liefert jetzt den StreamObserver zurück, mit dem man dann die Parameter übergeben kann: 
      StreamObserver<Foo> sender =
  	  stub.storeMeasurement(/* StreamObserver<Bar> */ receiver);
 
sender.onNext(/* ... */);
sender.onCompleted();
Protobuf Typendefinition

Siehe auch

  • Die Nummern beschreiben nur in welcher Reihenfolge die Daten geschickt werden, keine Semantische Bedeutung für den Typ selbst.
message Adress {
	string street = 1;
	uint32 plz = 2;
}
 
message User {
	string username = 1; 
	uint32 age = 2;
	Adress = 3;
	repeated Interest interests = 4; // repeated -> Array
}

Enums

enum Interest {
	WORKING = 0; 
	LEARNING = 1; 
	CHILLING = 2;
}

Externe Proto-Dateien importieren

Mit import können andere Proto-Dateien importiert werden. Google hat bereits einige oft verwendete Typen definiert:

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto"
stub.storeMeasurement(
	/* Request */ request,
	/* StreamObserver<Response> */ receiver
);

Der Funktionsaufruf erwartet 2 Parameter:

  • Ein Request-Wert (das Funktionsparameter), der an den Client geschickt wird
  • Ein StreamObserver, der die Response vom Server mit onNext asynchron verarbeitet.
public void dong(Request request, StreamObserver<Response> responseObserver)

Das Interface erwartet 2 Parameter:

  • Das gesendete Request-Objekt
  • Ein StreamObserver, über den man mit onNext eine Response senden kann.
StreamObserver<Foo> sender = 
	  stub.storeMeasurement(/* StreamObserver<Response> */ receiver);
	  
sender.onNext(/* ... */);
sender.onCompleted();
  • Wie gehabt in asynchronen Clients erwartet der Funktionsaufruf einen StreamObserver, der über onNext gestreamte Response-Werte verarbeitet.
  • Statt jedoch die Parameter beim Funktionsaufruf zu übergeben, gibt die Funktion einen StreamObserver zurück, mit dem man Request-Objekte streamen kann.
public StreamObserver<Request> dong(StreamObserver<Response> responseObserver)
  • Als Parameter bekommt die Funktion wie gehabt den StreamObserver, mit der man Response-Werte zurück an den Client streamen kann.
  • Die Funktion muss jetzt aber statt void einen StreamObserver zurückgeben, in dem onNext usw. implementiert werden, welche die vom Client gestreamten Request-Parameter verarbeiten.

Server

Client

Kein Parameter-Streaming

Parameter-Streaming

Protobuf

Protobuf ist eine Sprache zur Definition von Schnittstellen für gRPC.

Aufbau

syntax = "proto3";
 
// Typen- und Interfacedefinitionen hier...

Kompilieren

Diese Datei wird zur Designzeit vom Compiler ptotoc übersetzt, hier z.B. für Java:

protoc --java_out=./target ./src/service.proto