Vert.x gRPC 模块

<plugin>
    <groupId>io.github.ascopes</groupId>
    <artifactId>protobuf-maven-plugin</artifactId>
    <version>${ascopes.grpc.version}</version>
    <configuration>
        <protocVersion>${protobuf.version}</protocVersion>
    </configuration>
    <executions>
        <execution>
            <id>compile-java</id>
            <configuration>
                <jvmMavenPlugins>
                    <jvmMavenPlugin>
                        <groupId>io.vertx</groupId>
                        <artifactId>vertx-grpc-protoc-plugin2</artifactId>
                        <version>${vertx.version}</version>
                        <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
                        <options>grpc-client=true,grpc-service=true,grpc-io=true</options>
                        <!-- 或者你也可以使用 jvmArgs -->
                        <jvmArgs>
                            <jvmArg>--grpc-client=true</jvmArg>
                            <jvmArg>--grpc-service=true</jvmArg>
                            <jvmArg>--grpc-io=true</jvmArg>
                        </jvmArgs>
                    </jvmMavenPlugin>
                </jvmMavenPlugins>
            </configuration>
        </execution>
    </executions>
</plugin>

<plugin>
    <groupId>org.xolstice.maven.plugins</groupId>
    <artifactId>protobuf-maven-plugin</artifactId>
    <version>0.6.1</version>
    <configuration>
        <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
        <pluginId>grpc-java</pluginId>
        <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
        <protocPlugins>
            <protocPlugin>
                <id>vertx-grpc-protoc-plugin2</id>
                <groupId>io.vertx</groupId>
                <artifactId>vertx-grpc-protoc-plugin2</artifactId>
                <version>${stack.version}</version>
                <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
                <args>
                    <arg>--grpc-client=true</arg>
                    <arg>--grpc-service=true</arg>
                    <arg>--grpc-io=true</arg>
                </args>
            </protocPlugin>
        </protocPlugins>
    </configuration>
    <executions>
        <execution>
            <id>compile</id>
            <configuration>
                <outputDirectory>${project.basedir}/src/main/java</outputDirectory>
                <clearOutputDirectory>false</clearOutputDirectory>
            </configuration>
            <goals>
                <goal>compile</goal>
                <goal>compile-custom</goal>
            </goals>
        </execution>
    </executions>
</plugin>

<build>
  ...
  <extensions>
    <extension>
      <groupId>kr.motd.maven</groupId>
      <artifactId>os-maven-plugin</artifactId>
      <version>1.4.1.Final</version>
    </extension>
  </extensions>
  ...
</build>

<jvmMavenPlugin>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin2</artifactId>
  <version>${vertx.version}</version>
  <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
  <options>grpc-client=true,grpc-service=true,grpc-io=true</options>
</jvmMavenPlugin>

...
apply plugin: 'com.google.protobuf'
...
buildscript {
  ...
  dependencies {
    // 假设使用 Gradle 2.12 或更高版本。若为更早的 Gradle 版本，请使用插件版本 0.7.5
    classpath 'com.google.protobuf:protobuf-gradle-plugin:0.8.0'
  }
}
...
protobuf {
  protoc {
    artifact = 'com.google.protobuf:protoc:3.2.0'
  }
  plugins {
    grpc {
      artifact = "io.grpc:protoc-gen-grpc-java:1.25.0"
    }
    vertx {
      artifact = "io.vertx:vertx-grpc-protoc-plugin2:${vertx.grpc.version}"
    }
  }
  generateProtoTasks {
    all()*.plugins {
      grpc
      vertx
    }
  }
}

GrpcServer grpcServer = GrpcServer.server(vertx);

HttpServer server = vertx.createHttpServer(config, sslOptions);

server
  .requestHandler(grpcServer)
  .listen();

route.consumes("application/grpc").handler(rc -> grpcServer.handle(rc.request()));

public interface Greeter {

  Future<examples.grpc.HelloReply> sayHello(examples.grpc.HelloRequest request);

}

public class GreeterService implements Greeter {

  public Future<examples.grpc.HelloReply> sayHello(examples.grpc.HelloRequest request) {
    // ...
  }

  protected void sayHello(examples.grpc.HelloRequest request, Completable<examples.grpc.HelloReply> response) {
    // ...
  }
}

GreeterGrpcService service = GreeterGrpcService.create(new GreeterService() {
  @Override
  public Future<HelloReply> sayHello(HelloRequest request) {
    return Future.succeededFuture(HelloReply.newBuilder()
      .setMessage("Hello " + request.getName())
      .build());
  }
});

grpcServer.add(service);

request.pause();

performAsyncOperation().onComplete(ar -> {
  // 然后恢复
  request.resume();
});

if (response.writeQueueFull()) {
  response.drainHandler(v -> {
    // 再次可写
  });
} else {
  response.write(item);
}

long timeout = request.timeout();

if (timeout > 0L) {
  // 已接收到超时信息
}

GrpcServer server = GrpcServer.server(vertx, new GrpcServerOptions()
  .setScheduleDeadlineAutomatically(true)
  .setDeadlinePropagation(true)
);

ServiceMethod<JsonObject, JsonObject> sayHello = ServiceMethod.server(
  ServiceName.create("helloworld", "Greeter"),
  "SayHello",
  GrpcMessageEncoder.JSON_OBJECT,
  GrpcMessageDecoder.JSON_OBJECT
);

server.callHandler(sayHello, request -> {
  request.last().onSuccess(helloRequest -> {
    request.response().end(new JsonObject().put("message", "Hello " + helloRequest.getString("name")));
  });
});

if (response.acceptedEncodings().contains("gzip")) {
  response.encoding("gzip");
}

// 在定义编码后写入条目
response.write(Item.newBuilder().setValue("item-1").build());
response.write(Item.newBuilder().setValue("item-2").build());
response.write(Item.newBuilder().setValue("item-3").build());

ServiceName greeterServiceName = ServiceName.create("helloworld", "Greeter");

server.callHandler(request -> {

  if (request.serviceName().equals(greeterServiceName) && request.methodName().equals("SayHello")) {

    request.handler(protoHello -> {
      // 处理 Protobuf 编码的 hello
      performAsyncOperation(protoHello)
        .onSuccess(protoReply -> {
          // 以 Protobuf 编码的 reply 进行回复
          request.response().end(protoReply);
        }).onFailure(err -> {
          request.response()
            .status(GrpcStatus.ABORTED)
            .end();
        });
    });
  } else {
    request.response()
      .status(GrpcStatus.NOT_FOUND)
      .end();
  }
});

ServiceName greeterServiceName = ServiceName.create("helloworld", "Greeter");

server.callHandler(request -> {

  if (request.serviceName().equals(greeterServiceName) && request.methodName().equals("SayHello")) {

    request.messageHandler(helloMessage -> {

      // 可以是 identity 或 gzip
      String helloEncoding = helloMessage.encoding();

      // 处理 hello 消息
      handleGrpcMessage(helloMessage)
        .onSuccess(replyMessage -> {
          // 回复 reply 消息

          // 可以是 identity 或 gzip
          String replyEncoding = replyMessage.encoding();

          // 发送回复
          request.response().endMessage(replyMessage);
        }).onFailure(err -> {
          request.response()
            .status(GrpcStatus.ABORTED)
            .end();
        });
    });
  } else {
    request.response()
      .status(GrpcStatus.NOT_FOUND)
      .end();
  }
});

GrpcServer grpcServer = GrpcServer.server(vertx);

// 添加反射服务
grpcServer.addService(ReflectionService.v1());

GreeterGrpcService greeterService = new GreeterGrpcService() {
  @Override
  public Future<HelloReply> sayHello(HelloRequest request) {
    return Future.succeededFuture(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
  }
};

grpcServer.addService(greeterService);

// 启动服务端
vertx.createHttpServer(config, sslOptions)
  .requestHandler(grpcServer)
  .listen();

GrpcServer grpcServer = GrpcServer.server(vertx);

// 创建健康检查服务实例
HealthService healthService = HealthService.create(vertx);

// 为你的服务注册健康检查
healthService.register("my.service.name", () -> Future.succeededFuture(true));

// 将健康检查服务添加到 gRPC 服务端
grpcServer.addService(healthService);

// 启动服务端
vertx.createHttpServer(config, sslOptions)
  .requestHandler(grpcServer)
  .listen();

GrpcClient client = GrpcClient.client(vertx);

public interface Greeter {

  Future<examples.grpc.HelloReply> sayHello(examples.grpc.HelloRequest request);

}

@io.vertx.codegen.annotations.VertxGen
public interface GreeterClient extends Greeter {

  @io.vertx.codegen.annotations.GenIgnore(io.vertx.codegen.annotations.GenIgnore.PERMITTED_TYPE)
  Future<examples.grpc.HelloReply> sayHello(examples.grpc.HelloRequest request);
}

GreeterGrpcClient greeterClient = GreeterGrpcClient.create(client, SocketAddress.inetSocketAddress(port, host));

<protocPlugin>
  <id>vertx-grpc-protoc-plugin2</id>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin2</artifactId>
  <version>${stack.version}</version>
  <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
  <args>
    <arg>--grpc-client</arg>
    <arg>--vertx-codegen</arg>
  </args>
</protocPlugin>

@io.vertx.codegen.annotations.VertxGen
public interface GreeterClient {
  ...
}

<plugin>
  <artifactId>maven-compiler-plugin</artifactId>
  <executions>
    <execution>
      <id>default-compile</id>
      <configuration>
        <annotationProcessorPaths>
          <annotationProcessorPath>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-codegen</artifactId>
            <classifier>processor</classifier>
            <version>${vertx.version}</version>
          </annotationProcessorPath>
          <annotationProcessorPath>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-rx-java3-gen</artifactId>
            <version>${vertx.version}</version>
          </annotationProcessorPath>
        </annotationProcessorPaths>
      </configuration>
    </execution>
  </executions>
</plugin>

// 使用 RxJava 版本
GreeterClient client = io.grpc.examples.rxjava3.helloworld.GreeterClient.create(grpcClient, SocketAddress.inetSocketAddress(8080, "localhost"));

// 获取 Single 而不是 Future
Single<HelloReply> reply = client.sayHello(HelloRequest.newBuilder().setName("World").build());

Flowable<HelloRequest> requestStream = Flowable.just("World", "Monde", "Mundo")
      .map(name -> HelloRequest.newBuilder().setName(name).build());

// 使用 Flowable 代替 Vert.x Stream
Flowable<String> responseStream = client
  .sayHelloStreaming(stream)
  .map(HelloReply::getMessage);

if (request.writeQueueFull()) {
  request.drainHandler(v -> {
    // 再次可写
  });
} else {
  request.write(item);
}

response.pause();

performAsyncOperation().onComplete(ar -> {
  // 然后恢复
  response.resume();
});

GrpcClient client = GrpcClient.client(vertx, new GrpcClientOptions()
  .setTimeout(10)
  .setTimeoutUnit(TimeUnit.SECONDS)
  .setScheduleDeadlineAutomatically(true));

Future<GrpcClientRequest<HelloRequest, HelloReply>> fut = client.request(server, GreeterGrpcClient.SayHello);
fut.onSuccess(request -> {

  request
    // 针对该请求设置 10 秒超时，该超时信息将发送给 gRPC 服务
    .timeout(10, TimeUnit.SECONDS);

  request.end(HelloRequest.newBuilder().setName("Bob").build());
});

request.cancel();

client
  .request(server, GreeterGrpcClient.SayHello).compose(request -> {
    request.format(WireFormat.JSON);
    request.end(HelloRequest
      .newBuilder()
      .setName("Bob")
      .build());
    return request.response().compose(response -> response.last());
  }).onSuccess(reply -> {
    System.out.println("Received " + reply.getMessage());
  });

ServiceMethod<JsonObject, JsonObject> sayHello = ServiceMethod.client(
  ServiceName.create("helloworld", "Greeter"),
  "SayHello",
  GrpcMessageEncoder.JSON_OBJECT,
  GrpcMessageDecoder.JSON_OBJECT
);
client
  .request(server, sayHello).compose(request -> {
    request.end(new JsonObject().put("name", "Bob"));
    return request.response().compose(response -> response.last());
  }).onSuccess(reply -> {
    System.out.println("Received " + reply.getString("message"));
  });

GreeterGrpcClient greeterClient = GreeterGrpcClient.create(client, SocketAddress.inetSocketAddress(port, host), WireFormat.JSON);

request.encoding("gzip");

// 在定义编码后写入条目
request.write(Item.newBuilder().setValue("item-1").build());
request.write(Item.newBuilder().setValue("item-2").build());
request.write(Item.newBuilder().setValue("item-3").build());

GrpcIoServer grpcServer = GrpcIoServer.server(vertx);

GreeterGrpc.GreeterImplBase service = new GreeterGrpc.GreeterImplBase() {
  @Override
  public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
    responseObserver.onNext(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
    responseObserver.onCompleted();
  }
};

// 在 gRPC 服务端中绑定服务
grpcServer.addService(service);

// 启动 HTTP/2 服务端
vertx.createHttpServer(config)
  .requestHandler(grpcServer)
  .listen();

EventBusGrpcServer server = EventBusGrpcServer.server(vertx);

GrpcIoServiceBridge bridge = GrpcIoServiceBridge.bridge(service);

server.addService(bridge);

<protocPlugin>
  <id>vertx-grpc-protoc-plugin2</id>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin2</artifactId>
  <version>${stack.version}</version>
  <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
  <args>
    <arg>--grpc-io</arg>
  </args>
</protocPlugin>

GrpcIoServer grpcServer = GrpcIoServer.server(vertx);

BindableService service = GreeterGrpcIo.bindableServiceOf(new GreeterService() {
  @Override
  public Future<HelloReply> sayHello(HelloRequest request) {
    return Future.succeededFuture(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
  }
});

// 在 gRPC 服务端中绑定服务
grpcServer.addService(service);

// 启动 HTTP/2 服务端
vertx.createHttpServer(config, sslOptions)
  .requestHandler(grpcServer)
  .listen();

GrpcIoServer grpcServer = GrpcIoServer.server(vertx);

// 添加反射服务
grpcServer.addService(ReflectionService.v1());

GreeterGrpc.GreeterImplBase greeterService = new GreeterGrpc.GreeterImplBase() {
  @Override
  public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
    responseObserver.onNext(HelloReply.newBuilder().setMessage("Hello " + request.getName()).build());
    responseObserver.onCompleted();
  }
};

// 在 gRPC 服务端中绑定服务
grpcServer.addService(greeterService);

// 启动 HTTP/2 服务端
vertx.createHttpServer(config, sslOptions)
  .requestHandler(grpcServer)
  .listen();

grpcurl -plaintext localhost:50051 list

grpcurl -plaintext localhost:50051 describe .helloworld.HelloRequest

grpcurl -plaintext -d '{"name": "Vert.x"}' localhost:50051 helloworld.Greeter

GrpcIoClientChannel channel = new GrpcIoClientChannel(client, SocketAddress.inetSocketAddress(443, "example.com"));

GreeterGrpc.GreeterStub greeter = GreeterGrpc.newStub(channel);

StreamObserver<HelloReply> observer = new StreamObserver<>() {
  @Override
  public void onNext(HelloReply value) {
    // 处理响应
  }

  @Override
  public void onCompleted() {
    // 完成
  }

  @Override
  public void onError(Throwable t) {
    // 发生异常
  }
};

greeter.sayHello(HelloRequest.newBuilder().setName("Bob").build(), observer);

EventBusGrpcClient client = EventBusGrpcClient.client(vertx);

GrpcIoClientChannel channel = new GrpcIoClientChannel(client);

GreeterGrpc.GreeterStub greeter = GreeterGrpc.newStub(channel);

GreeterGrpc.GreeterStub greeter = GreeterGrpc.newStub(channel).withDeadlineAfter(10, TimeUnit.SECONDS);

greeter.sayHello(HelloRequest.newBuilder().setName("Bob").build(), observer);

<protocPlugin>
  <id>vertx-grpc-protoc-plugin2</id>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-grpc-protoc-plugin2</artifactId>
  <version>${stack.version}</version>
  <mainClass>io.vertx.grpc.plugin.VertxGrpcGenerator</mainClass>
  <args>
    <arg>grpc-io</arg>
  </args>
</protocPlugin>

GrpcIoClientChannel channel = new GrpcIoClientChannel(client, SocketAddress.inetSocketAddress(443, "example.com"));

GreeterGrpcIo.GreeterStub greeter = GreeterGrpcIo.newStub(vertx, channel);

Future<HelloReply> future = greeter.sayHello(HelloRequest.newBuilder().setName("Bob").build());

future
  .onSuccess(reply -> {
    // 处理响应
  })
  .onFailure(err -> {
    // 发生异常
  });

<dependency>
 <groupId>io.vertx</groupId>
 <artifactId>vertx-grpc-context-storage</artifactId>
 <version>4.5.35</version>
</dependency>

dependencies {
 compile 'io.vertx:vertx-grpc-context-storage:4.5.35'
}