grpc-reactor 实战（三）：把 Unary RPC 映射到 Reactor Netty HTTP/2

协议值与消息帧完成后，Stage 2 实现了第一条端到端调用：plaintext h2c unary RPC。这部分已经进入 main，后续 Stage 3 和 Stage 4 继续在同一 transport primitive 上完成了全部四种 RPC cardinality。

Method descriptor 是协议与类型的交点

一个方法需要精确的 service name、method name、cardinality，以及 request/response marshaller：

var echo = new GrpcMethod<>(
    "testing.EchoService",
    "Echo",
    GrpcMethod.Cardinality.UNARY,
    new ProtobufMarshaller<>(StringValue.parser()),
    new ProtobufMarshaller<>(StringValue.parser()));

它生成的 path 必须是：

1	/testing.EchoService/Echo

服务注册表按完整 path 精确匹配。未知 path 返回 UNIMPLEMENTED，重复注册同一路径在构建 service definition 时立即失败。

Server 先验证协议，再订阅业务

ReactorGrpcServer 使用 Reactor Netty h2c：

DisposableServer bound = HttpServer.create()
    .host(host)
    .port(port)
    .protocol(HttpProtocol.H2C)
    .handle(handler::handle)
    .bindNow(Duration.ofSeconds(10));

请求进入后按顺序验证：

HTTP method 必须是 POST；
content-type 必须是 application/grpc 或 application/grpc+...；
te 必须声明 trailers；
path 必须存在；
当前只允许 unary cardinality；
metadata 与 message size 不能超过限制。

验证通过后才创建 GrpcCallContext 并订阅 request body，避免无效请求进入业务 Handler。

HTTP 200 不代表 RPC 成功

Server 先写 compatible content-type，最终状态由 trailing headers 产生：

response.status(200)
    .header(HttpHeaderNames.CONTENT_TYPE, "application/grpc+proto");

response.trailerHeaders(trailers -> {
  GrpcException error = terminal.get();
  if (error == null) {
    writeStatus(trailers, GrpcStatus.OK, GrpcMetadata.empty());
  } else {
    writeStatus(trailers, error.status(), error.trailers());
  }
});

应用抛出的 GrpcException 保留明确 status 与 trailing metadata。普通应用异常映射为 UNKNOWN；协议违规映射为 INTERNAL。Fatal JVM error 不应该被包装成普通业务状态。

Client 收到响应后必须读取最终 grpc-status。HTTP 200 却没有 status 是 INTERNAL；非 gRPC HTTP 404 按标准 fallback 映射为 UNIMPLEMENTED，503 则映射为 UNAVAILABLE。

Metadata 必须保序并允许重复

gRPC metadata 不是普通 Map<String, String>。同一个 key 可以重复，顺序需要保留，-bin 后缀表示二进制值并在 wire 上使用无 padding Base64。

GrpcMetadata 因此内部保存 immutable entry list：

GrpcMetadata.builder()
    .addAscii("trace-id", "abc")
    .addBinary("span-bin", bytes)
    .build();

用户 metadata 不能覆盖 content-type、te、grpc-status、grpc-timeout 和 HTTP/2 pseudo headers 等保留字段。读取时还要限制编码后总大小，防止 peer 用 headers 消耗无界内存。

Deadline 的 wire value 不能简单截断

grpc-timeout 由最多八位数字和一个单位组成，例如 100m、2S。格式化时必须向上取整，确保 wire deadline 不会比调用者要求更短：

BigInteger amount = ceilDivide(nanos, unit.nanos);
if (amount.compareTo(MAX_VALUE) <= 0) {
  return amount + String.valueOf(unit.suffix);
}

在本篇对应的 Stage 2 提交中，client 只会写入 timeout header，并把 Reactor Netty response timeout 设置得略长，用于区分 gRPC deadline 与底层连接等待。仅能解析 header 还不算完整支持；Stage 5 后续补上了 client 与 server 的 Reactor deadline 定时器、handler 取消和 GrpcCallContext 传播。当前实现仍不会替业务代码自动为 nested call 派生更短的 deadline。

Unary cardinality 用 single 强制执行

Client 对 unary request 使用 requests.single()，response 同样要求一个值。空请求、两个请求、空响应和两个响应都不是隐式默认值，而是明确错误。

当前 transport 测试覆盖：

unary message 与 ASCII metadata 往返；
declared error 和 trailing metadata；
unknown method；
空/重复 unary 数据；
非 gRPC HTTP 响应与缺失 status；
client cancellation 传播到 server publisher；
connection failure 映射为 UNAVAILABLE。

Unary vertical slice 的测试继续作为 streaming transport 的回归基线。Server、client 与 bidirectional streaming 已在后续提交完成；TLS、GOAWAY、keepalive、compression、deadline、连接池和 transport diagnostics 也已在 Stage 5 完成并进入 main。

Client 请求的完整生命周期

ReactorGrpcClient 的 unary 调用最终映射为一次完整的 HTTP/2 stream 交互。下面是从用户调用到网络响应的全路径：

client.unary(method, Mono.just(request))
    │
    ├─ Flux.defer() → 惰性组装，每次订阅独立
    │
    ├─ channel.select() → 选取 subchannel（Stage 2 固定单连接）
    │
    ├─ httpClient.post(method.path())
    │   ├─ 写 HTTP/2 request headers
    │   │   content-type: application/grpc+proto
    │   │   te: trailers
    │   │   grpc-accept-encoding: gzip
    │   │   grpc-timeout: <if set>
    │   │   custom metadata
    │   │
    │   ├─ onEventLoop() → 在连接的 event loop 上序列化
    │   │
    │   ├─ GrpcOutboundMessageEncoder.encode()
    │   │   ├─ marshaller.serialize() → Protobuf → ByteBuf
    │   │   ├─ 校验 size ≤ maxOutboundMessageSize
    │   │   └─ GrpcFrameCodec.encode() → 5 字节 header + payload
    │   │
    │   └─ outbound.send(requestFrames) → HTTP/2 DATA frame
    │
    └─ .response((response, content) → decodeResponse)
        ├─ 检查 response headers 中的 grpc-status（trailers-only 错误）
        ├─ 校验 HTTP 200 + content-type = application/grpc
        ├─ 读取 grpc-encoding header
        ├─ GrpcFrameCodec.decode(content) → 增量帧解析
        ├─ marshaller.deserialize() → ByteBuf → Response protobuf
        └─ 读取 trailers 中的 grpc-status + grpc-message

关键设计：整个 pipeline 是 lazy 的——在 Mono 被订阅之前不会发起 HTTP/2 连接。这与 grpc-java 的 ClientCall.start() 即触发连接的模式不同。

为什么要在 event loop 上序列化

Reactor Netty 的每条 HTTP/2 连接绑定到一个 Netty event loop 线程。如果 marshalling 发生在用户线程，序列化产生的 ByteBuf 需要跨线程传递给 event loop，带来额外的同步开销。

onEventLoop() 把序列化推迟到连接的 event loop 上执行：

private <T> Flux<T> onEventLoop(NettyOutbound outbound, Flux<T> source) {
    return source.publishOn(
        Schedulers.fromExecutor(outbound.alloc().buffer(0).alloc()
            ... // 获取 event loop executor
        ));
}

这样做的好处：

序列化和网络写入在同一线程，无锁；
ByteBuf 的 allocator 与连接一致，减少跨 arena 分配；
concatMap(..., 1) 控制 prefetch，保证消息严格按序发出。

错误映射：从 HTTP/2 异常到 gRPC 状态

网络层的错误不能直接暴露给用户，需要映射到 gRPC 语义：

网络异常	gRPC Status	含义
`Http2Exception.StreamException(REFUSED_STREAM)`	UNAVAILABLE	Server 拒绝了这条 stream，可安全重试
GOAWAY frame	UNAVAILABLE	Server 正在优雅关闭
连接关闭 / reset	UNAVAILABLE	网络断开
HTTP 404（非 gRPC 响应）	UNIMPLEMENTED	路径不存在（可能经过了非 gRPC 代理）
HTTP 503	UNAVAILABLE	服务暂时不可用
缺失 `grpc-status`	INTERNAL	协议违规
`content-type` 不匹配	INTERNAL	响应不是 gRPC

这个映射遵循 gRPC 协议规范中定义的 HTTP status code 到 gRPC status 的转换表。UNAVAILABLE 意味着客户端可以重试；INTERNAL 和 UNIMPLEMENTED 通常不值得重试。

Client 取消如何传播到 Server

Reactor 的 dispose() 需要映射为 HTTP/2 的 RST_STREAM，让 server 知道这条调用已经不需要了：

client: subscription.dispose()
    → Reactor Netty 观察到 subscriber cancel
    → 发送 HTTP/2 RST_STREAM frame
    → Server 连接收到 RST_STREAM
    → request body publisher 被 cancel
    → server handler 的 Mono/Flux 收到 cancel signal

Server 侧通过 connection handler 监听 stream 关闭事件，将其转换为 peerCancellation signal。业务 handler 使用 takeUntilOther(peerCancellation) 订阅这个 signal，确保 client 取消后 server 不再做无用计算。

这一机制让 Reactor 的背压语义（cancel = 我不再需要数据）与 gRPC 的取消语义（RST_STREAM = 调用已中止）自然对齐。

与 grpc-java 的互操作验证

Stage 2 的退出条件不是”自己调自己成功”，而是必须与 grpc-java 双向互通。UnaryInteroperabilityTest 验证两个方向：

Reactor client → grpc-java server：

@Test
void reactorClientCallsGrpcJavaServer() {
    try (var fixture = GrpcJavaFixture.start();
         var client = ReactorGrpcClient.builder().port(fixture.port()).build()) {
        var response = stub.unary(Mono.just(request("hello"))).block();
        assertEquals("hello", response.getValue());

        // 验证错误 status 和 trailing metadata 正确传递
        var error = assertThrows(GrpcException.class,
            () -> stub.unary(Mono.just(request("error"))).block());
        assertEquals(GrpcStatus.Code.INVALID_ARGUMENT, error.status().code());
        assertEquals("rejected", error.status().message());
        assertEquals("fixture", error.trailers().getLast("error-id").orElseThrow().asciiValue());
    }
}

grpc-java client → Reactor server：

@Test
void grpcJavaClientCallsReactorServer() {
    var service = InteropTestServiceReactorGrpc.bindService(new Service() {
        @Override
        public Mono<TestResponse> unary(GrpcCallContext context, Mono<TestRequest> requests) {
            return requests.map(req -> {
                if (req.getValue().equals("error")) {
                    throw new GrpcException(
                        new GrpcStatus(GrpcStatus.Code.INVALID_ARGUMENT, "rejected"),
                        GrpcMetadata.builder().addAscii("error-id", "reactor").build());
                }
                return TestResponse.newBuilder().setValue(req.getValue()).build();
            });
        }
    });

    try (var server = ReactorGrpcServer.builder().addService(service).start()) {
        var stub = InteropTestServiceGrpc.newBlockingStub(channel);
        assertEquals("hello", stub.unary(request("hello")).getValue());

        var error = assertThrows(StatusRuntimeException.class,
            () -> stub.unary(request("error")));
        assertEquals(Status.Code.INVALID_ARGUMENT, error.getStatus().getCode());
    }
}

两个方向都验证了：正常响应、错误 status code、错误 message 的 percent-encoding 往返，以及自定义 trailing metadata 的传递。只有 grpc-java 能正确解析 Reactor server 的响应，才能证明 wire protocol 实现是兼容的。

Stage 2 退出条件

./gradlew clean spotlessCheck test --no-daemon 全量通过
Reactor client ↔ grpc-java server 双向 unary 调用成功
grpc-java client ↔ Reactor server 双向 unary 调用成功
错误 status、message、trailing metadata 正确传递
Client cancellation 通过 RST_STREAM 传播到 server
连接失败映射为 UNAVAILABLE
未知路径映射为 UNIMPLEMENTED
Stage 0 和 Stage 1 的全部测试继续绿灯

下一篇进入 Stage 3 和 Stage 4：在同一 transport primitive 上实现 server streaming、client streaming 和 bidirectional streaming，面对的新挑战是多消息流的背压协调、有界入站缓冲和流提前终止的资源回收。