事件总线跨端桥接

Vert.x gRPC EventBus 网桥

Preview

Vert.x gRPC EventBus 网桥（Bridge）是连接 gRPC 与 Vert.x EventBus 的桥梁。

要在项目中使用此组件，请在构建描述文件的 dependencies 配置节中添加以下依赖：

Maven

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-eventbus-bridge-grpc</artifactId>
  <version>5.2.1</version>
</dependency>

Gradle

compile 'io.vertx:vertx-eventbus-bridge-grpc:5.2.1'

gRPC EventBus 网桥构建于 gRPC 之上，这意味着任何能够创建 gRPC 客户端的应用程序，都可以通过事件总线与远端的 Vert.x 实例进行交互。gRPC 网桥的主要使用场景是为需要标准化、高性能且跨语言方式与 Vert.x EventBus 通信的应用程序提供支持。

gRPC 提供了一个现代、开源的 RPC 框架，采用 HTTP/2 作为传输层，并使用 Protocol Buffers 作为接口定义语言（IDL）。

该网桥还支持 gRPC-Web，允许 Web 浏览器客户端直接与 gRPC 服务器通信。

创建 gRPC EventBus 网桥服务端

你可以按如下方式创建 gRPC EventBus 网桥服务端：

GrpcBridgeOptions options = new GrpcBridgeOptions()
  .addInboundPermitted(new PermittedOptions().setAddress("hello"))
  .addInboundPermitted(new PermittedOptions().setAddress("echo"))
  .addOutboundPermitted(new PermittedOptions().setAddress("news"));

// 创建网桥
GrpcEventBusBridge bridge = GrpcEventBusBridge.create(vertx, options);

// 创建 gRPC 服务端
GrpcServer grpcServer = GrpcServer.server(vertx).addService(bridge);

// 添加网桥服务
grpcServer.addService(bridge);

HttpServer server = vertx
  .createHttpServer(new HttpServerOptions()
    .setSsl(true)
    .setKeyCertOptions(new JksOptions()
      .setPath("/path/to/keycert.jks")
      .setPassword("the-password"))
  )
  .requestHandler(grpcServer);

// 启动网桥
server.listen(443).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("gRPC EventBus Bridge started");
  } else {
    System.err.println("Failed to start gRPC EventBus Bridge: " + ar.cause());
  }
});

创建网桥涉及以下步骤：

配置合法的入站/出站权限规则并创建网桥；
创建 gRPC 服务端；
将网桥注册到 gRPC 服务端中；
创建托管 gRPC 服务端的 HTTP 服务器并监听端口。

你可以查阅 Vert.x gRPC 文档以及 Vert.x Core 文档以了解有关 Vert.x gRPC 与 HTTP 服务器的更多细节。

创建 Vert.x gRPC 客户端

以下演示如何创建用于连接网桥的 gRPC 客户端：

GrpcClient client = GrpcClient.client(vertx);
SocketAddress socketAddress = SocketAddress.inetSocketAddress(7000, "localhost");
EventBusBridgeGrpcClient bridgeClient = EventBusBridgeGrpcClient.create(client, socketAddress);

与网桥交互

本节介绍使用 Vert.x gRPC 客户端与 gRPC 网桥交互的核心通信模式。

这些交互模式本身并不绑定 Vert.x gRPC 客户端。你可以将它们应用于由 eventbus.proto 定义的 EventBus RPC 与消息结构的任意原生 gRPC 客户端实现中。

发送消息

向指定地址发送点对点消息：

JsonObject message = new JsonObject().put("value", "Hello from gRPC client");

// 转换为 Protobuf 结构
JsonValue messageBody = JsonValue.newBuilder().setText(message.encode()).build();

// 构建请求
SendOp request = SendOp.newBuilder()
  .setAddress("hello")
  .setBody(messageBody)
  .build();

// 发送消息
grpcClient.send(request).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Message sent successfully");
  } else {
    System.err.println("Failed to send message: " + ar.cause());
  }
});

请求-响应模式

发送请求并等待接收响应结果：

JsonObject message = new JsonObject().put("value", "Hello from gRPC client");

// 转换为 JsonValue
JsonValue messageBody = JsonValue.newBuilder().setText(message.encode()).build();

// 构建带超时的请求
RequestOp request = RequestOp.newBuilder()
  .setAddress("hello")
  .setBody(messageBody)
  .setReplyBodyFormatValue(JsonValueFormat.text_VALUE) // 请求编码后的字符串格式
  .setTimeout(Duration.newBuilder().setSeconds(10).build())  // 10 秒超时
  .build();

// 发送请求
grpcClient.request(request).onComplete(ar -> {
  if (ar.succeeded()) {
    EventBusMessage response = ar.result();
    // 将 Protobuf 结构解码为 JsonObject
    Object responseBody = Json.decodeValue(response.getBody().getText());
    System.out.println("Received response: " + responseBody);
  } else {
    System.err.println("Request failed: " + ar.cause());
  }
});

发布消息

向所有订阅者广播发布消息：

JsonObject message = new JsonObject().put("value", "Broadcast message");

// 转换为 JsonValue
JsonValue messageBody = JsonValue.newBuilder().setText(message.encode()).build();

// 构建请求
PublishOp request = PublishOp.newBuilder()
  .setAddress("news")
  .setBody(messageBody)
  .build();

// 发布消息
grpcClient.publish(request).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Message published successfully");
  } else {
    System.err.println("Failed to publish message: " + ar.cause());
  }
});

订阅消息

订阅来自指定地址的消息流：

SubscribeOp request = SubscribeOp.newBuilder()
  .setAddress("news")
  .setMessageBodyFormatValue(JsonValueFormat.text_VALUE) // 请求编码后的字符串格式
  .build();

// 订阅该地址
grpcClient.subscribe(request).onComplete(ar -> {
  if (ar.succeeded()) {
    // 获取数据流
    ReadStream<EventBusMessage> stream = ar.result();

    // 设置接收消息的处理器
    stream.handler(message -> {
      // 保存 consumer ID 以便后续取消订阅
      String consumerId = message.getConsumerId();

      // 将 Protobuf 结构解码为 JsonObject
      Object messageBody = Json.decodeValue(message.getBody().getText());
      System.out.println("Received message: " + messageBody);
    });

    // 异常处理
    stream.exceptionHandler(err -> {
      System.err.println("Stream error: " + err.getMessage());
    });
  } else {
    System.err.println("Failed to subscribe: " + ar.cause());
  }
});

取消订阅消息

取消对指定地址的订阅：

UnsubscribeOp request = UnsubscribeOp.newBuilder()
  .setConsumerId(consumerId)  // 订阅时接收到的 Consumer ID
  .build();

// 取消订阅
grpcClient.unsubscribe(request).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Unsubscribed successfully");
  } else {
    System.err.println("Failed to unsubscribe: " + ar.cause());
  }
});

回复消息

当 EventBusMessage 携带回复地址（replyAddress）时，表明该消息的发送方期待一个回复。

可以使用该回复地址将结果回传给发送者：

streamOfMessages.handler(message -> {

  String replyAddress = message.getReplyAddress();

  // 回复发送方
  SendOp reply = SendOp.newBuilder()
    .setAddress(replyAddress)
    .setBody(message.getBody())
    .build();

  // 发送回执消息
  grpcClient.send(reply);
});

健康检查

执行网桥健康检查：

grpcClient.ping(Empty.getDefaultInstance()).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Bridge is healthy");
  } else {
    System.err.println("Bridge health check failed: " + ar.cause());
  }
});

网桥事件监听

GrpcEventBusBridge 支持自定义网桥事件处理器，以便记录网桥操作日志或执行自定义访问控制：

GrpcEventBusBridge bridgeService = GrpcEventBusBridge.builder(vertx)
  .with(bridgeOptions)
  .withEventHandler(bridgeEvent -> {

    // 高级网桥事件处理
    switch (bridgeEvent.type()) {
      case SOCKET_CREATED:
        System.out.println("New gRPC client connected");
        break;
      case SOCKET_CLOSED:
        System.out.println("gRPC client disconnected");
        break;
      case SEND:
        System.out.println("Message sent to: " + bridgeEvent.getRawMessage().getString("address"));
        break;
      case PUBLISH:
        System.out.println("Message published to: " + bridgeEvent.getRawMessage().getString("address"));
        break;
      case RECEIVE:
        System.out.println("Message received from: " + bridgeEvent.getRawMessage().getString("address"));
        break;
      case REGISTER:
        System.out.println("Client registered for: " + bridgeEvent.getRawMessage().getString("address"));
        break;
      case UNREGISTER:
        System.out.println("Client unregistered from: " + bridgeEvent.getRawMessage().getString("address"));
        break;
    }

    // 务必完成事件以允许其继续流转
    bridgeEvent.complete(true);
  })
  .build();

协议规范说明

gRPC EventBus 网桥使用 Protocol Buffers 来定义服务与消息结构。服务定义包含以下核心操作：

Publish：向指定地址发布广播消息（单向通信）；
Send：向指定地址发送点对点消息；
Request：实现请求-响应（Request-Reply）通信模式；
Subscribe：订阅指定地址的消息（返回消息流）；
Unsubscribe：取消对指定地址的订阅；
Ping：健康检查探针。

消息结构定义如下：

JsonValue

网桥可以处理 JSON 值，取值范围包括 对象 (object)、数组 (array)、数值 (number)、字符串 (string)、true/false 或 null。 JsonValue 消息类型即专为此用途设计。

一个 JsonValue 可以包装字符串、字节数组或 google.protobuf.Value。

你可以从字符串创建 JsonValue：

JsonValue jsonValue = JsonValue.newBuilder().setText("4").build();

JsonObject json = new JsonObject().put("name", "Julien");
jsonValue = JsonValue.newBuilder().setText(json.encode()).build();

或者从字节数组创建：

JsonValue jsonValue = JsonValue.newBuilder().setBinary(ByteString.copyFromUtf8("4")).build();

JsonObject json = new JsonObject().put("name", "Julien");
jsonValue = JsonValue.newBuilder().setBinary(ByteString.copyFromUtf8(json.encode())).build();

或者从 google.protobuf.Value 创建：

JsonValue jsonValue = JsonValue.newBuilder().setProto(
  Value.newBuilder().setNumberValue(4)).build();

jsonValue = JsonValue.newBuilder().setProto(
  Value.newBuilder().setStructValue(Struct
    .newBuilder()
    .putFields("name", Value.newBuilder().setStringValue("Julien").build()))
).build();

你可以使用上述任意一种格式发送 JsonValue。

在与网桥交互时，作为客户端可以在通信协议中明确指定你期望接收的数据格式。

PublishOp

用于向指定地址发布广播消息：

message PublishOp {
  // 消息发送的目标地址
  string address = 1;

  // 消息头（Headers）
  map<string, string> headers = 2;

  // 消息负载（Payload）
  JsonValue body = 3;
}

SendOp

用于向指定地址发送点对点消息：

message SendOp {
  // 消息发送的目标地址
  string address = 1;

  // 消息头（Headers）
  map<string, string> headers = 2;

  // 消息负载（Payload）
  JsonValue body = 3;

  // 超时时长
  google.protobuf.Duration timeout = 4;
}

RequestOp

用于执行请求-响应通信：

message RequestOp {
  // 消息发送的目标地址
  string address = 1;

  // 消息头（Headers）
  map<string, string> headers = 2;

  // 消息负载（Payload）
  JsonValue body = 3;

  // 期望的回复正文格式
  JsonValueFormat reply_body_format = 4;

  // 超时时长
  google.protobuf.Duration timeout = 5;
}

SubscribeOp

用于向指定地址订阅消息：

message SubscribeOp {
  // 待订阅的目标地址
  string address = 1;

  // 消息头（Headers）
  map<string, string> headers = 2;

  // 期望网桥推送的消息正文格式
  JsonValueFormat message_body_format = 3;
}

UnsubscribeOp

用于取消对指定地址的订阅：

message UnsubscribeOp {
  // 用于订阅管理的 Consumer ID
  string consumer_id = 1;
}

EventBusMessage

用于从 EventBus 接收推送的消息：

message EventBusMessage {
  // 消息发送的目标地址
  string address = 1;

  // 用于订阅管理的 Consumer ID
  string consumer_id = 2;

  // 可选的回复地址
  // 若存在则表示发送方期待在此回复地址上收到回执响应
  // 该回复地址应当用于 Send 或 Request RPC 中
  string reply_address = 3;

  // 消息头（Headers）
  map<string, string> headers = 4;

  // 消息负载（Payload）
  JsonValue body = 5;

  // 用于错误响应的可选状态码
  google.rpc.Status status = 6;
}

Camel Bridge

Zipkin