feat(grpc): reactive zero-copy bridge with backpressure and metadata propagation

- **Refactor to fully reactive I/O:** Removed the blocking `GrpcHandler` (which relied on `InputStream.readNBytes`) and promoted `UnifiedGrpcBridge` to the primary `Route.Handler`.
- **Zero-copy ByteBuffer pipeline:** Migrated `GrpcDeframer` and `GrpcRequestBridge` to consume `Flow.Subscriber<ByteBuffer>` instead of `byte[]`. This allows Netty, Undertow, and Jetty to pipe native socket buffers directly into the gRPC state machine without intermediate array allocations.
- **Backpressure integration:** Wired gRPC's `ClientCallStreamObserver.setOnReadyHandler` directly to Jooby's `Flow.Subscription`, ensuring the server only demands more data when the internal gRPC buffers are ready.
- **Context propagation:** Added parsing and propagation of the `grpc-timeout` header into gRPC `CallOptions` (deadlines).
- **Metadata propagation:** Added HTTP header to gRPC `Metadata` mapping via `ClientInterceptors`, including base64 decoding for `-bin` suffixed headers.
- **Server Reflection upgrade:** Swapped the deprecated `v1alpha` reflection service for the stable `ProtoReflectionServiceV1`.
- **Testing:** Added comprehensive unit tests for `GrpcDeframer` fragmentation/coalescing logic and `GrpcRequestBridge` backpressure state handling.

Author: jknack

modules/jooby-grpc/src/test/java/io/jooby/grpc/GrpcDeframerTest.java

@@ -0,0 +1,124 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.grpc;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class GrpcDeframerTest {
+
+  private GrpcDeframer deframer;
+  private List<byte[]> outputMessages;
+
+  @BeforeEach
+  public void setUp() {
+    deframer = new GrpcDeframer();
+    outputMessages = new ArrayList<>();
+  }
+
+  @Test
+  public void shouldParseSingleCompleteMessage() {
+    byte[] payload = "hello grpc".getBytes();
+    ByteBuffer frame = createGrpcFrame(payload);
+
+    deframer.process(frame, msg -> outputMessages.add(msg));
+
+    assertEquals(1, outputMessages.size());
+    assertArrayEquals(payload, outputMessages.get(0));
+  }
+
+  @Test
+  public void shouldParseFragmentedHeader() {
+    byte[] payload = "fragmented header".getBytes();
+    byte[] frame = createGrpcFrame(payload).array();
+
+    // Send the first 2 bytes of the header
+    deframer.process(ByteBuffer.wrap(frame, 0, 2), msg -> outputMessages.add(msg));
+    assertEquals(0, outputMessages.size(), "Should not emit message yet");
+
+    // Send the rest of the header and the payload
+    deframer.process(ByteBuffer.wrap(frame, 2, frame.length - 2), msg -> outputMessages.add(msg));
+
+    assertEquals(1, outputMessages.size());
+    assertArrayEquals(payload, outputMessages.get(0));
+  }
+
+  @Test
+  public void shouldParseFragmentedPayload() {
+    byte[] payload = "this is a very long payload that gets split".getBytes();
+    byte[] frame = createGrpcFrame(payload).array();
+
+    // Send the 5-byte header + first 10 bytes of payload (15 bytes total)
+    deframer.process(ByteBuffer.wrap(frame, 0, 15), msg -> outputMessages.add(msg));
+    assertEquals(0, outputMessages.size(), "Should not emit message until full payload arrives");
+
+    // Send the remainder of the payload
+    deframer.process(ByteBuffer.wrap(frame, 15, frame.length - 15), msg -> outputMessages.add(msg));
+
+    assertEquals(1, outputMessages.size());
+    assertArrayEquals(payload, outputMessages.get(0));
+  }
+
+  @Test
+  public void shouldParseMultipleMessagesInSingleBuffer() {
+    byte[] payload1 = "message 1".getBytes();
+    byte[] payload2 = "message 2".getBytes();
+
+    ByteBuffer frame1 = createGrpcFrame(payload1);
+    ByteBuffer frame2 = createGrpcFrame(payload2);
+
+    // Combine both frames into a single buffer
+    ByteBuffer combined = ByteBuffer.allocate(frame1.capacity() + frame2.capacity());
+    combined.put(frame1).put(frame2);
+    combined.flip();
+
+    deframer.process(combined, msg -> outputMessages.add(msg));
+
+    assertEquals(2, outputMessages.size());
+    assertArrayEquals(payload1, outputMessages.get(0));
+    assertArrayEquals(payload2, outputMessages.get(1));
+  }
+
+  @Test
+  public void shouldHandleZeroLengthPayload() {
+    byte[] payload = new byte[0];
+    ByteBuffer frame = createGrpcFrame(payload);
+
+    deframer.process(frame, msg -> outputMessages.add(msg));
+
+    assertEquals(1, outputMessages.size());
+    assertArrayEquals(payload, outputMessages.get(0));
+  }
+
+  @Test
+  public void shouldHandleExtremeFragmentationByteByByte() {
+    byte[] payload = "byte by byte".getBytes();
+    byte[] frame = createGrpcFrame(payload).array();
+
+    for (byte b : frame) {
+      deframer.process(ByteBuffer.wrap(new byte[] {b}), msg -> outputMessages.add(msg));
+    }
+
+    assertEquals(1, outputMessages.size());
+    assertArrayEquals(payload, outputMessages.get(0));
+  }
+
+  /** Helper to wrap a raw payload in the standard 5-byte gRPC framing. */
+  private ByteBuffer createGrpcFrame(byte[] payload) {
+    ByteBuffer buffer = ByteBuffer.allocate(5 + payload.length);
+    buffer.put((byte) 0); // Compressed flag
+    buffer.putInt(payload.length); // Length
+    buffer.put(payload); // Data
+    buffer.flip();
+    return buffer;
+  }
+}

modules/jooby-grpc/src/test/java/io/jooby/grpc/GrpcRequestBridgeTest.java

@@ -0,0 +1,124 @@
+/*
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.grpc;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.Flow.Subscription;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import io.grpc.stub.ClientCallStreamObserver;
+
+public class GrpcRequestBridgeTest {
+
+  private ClientCallStreamObserver<byte[]> grpcObserver;
+  private Subscription subscription;
+  private GrpcRequestBridge bridge;
+
+  @BeforeEach
+  @SuppressWarnings("unchecked")
+  public void setUp() {
+    grpcObserver = mock(ClientCallStreamObserver.class);
+    subscription = mock(Subscription.class);
+    bridge = new GrpcRequestBridge(grpcObserver);
+  }
+
+  @Test
+  public void shouldRequestInitialDemandOnSubscribe() {
+    bridge.onSubscribe(subscription);
+
+    // Verify gRPC readiness handler is registered
+    verify(grpcObserver).setOnReadyHandler(any(Runnable.class));
+
+    // Verify initial demand of 1 is requested from Jooby
+    verify(subscription).request(1);
+  }
+
+  @Test
+  public void shouldDelegateOnNextAndRequestMoreIfReady() {
+    bridge.onSubscribe(subscription);
+
+    // Reset the mock to clear the initial request(1) from onSubscribe
+    reset(subscription);
+
+    // Simulate gRPC being ready to receive more data
+    when(grpcObserver.isReady()).thenReturn(true);
+
+    // Send a complete gRPC frame: Compressed Flag (0) + Length (4) + Payload ("test")
+    byte[] payload = "test".getBytes();
+    ByteBuffer frame = ByteBuffer.allocate(5 + payload.length);
+    frame.put((byte) 0).putInt(payload.length).put(payload).flip();
+
+    bridge.onNext(frame);
+
+    // Verify the deframed payload was passed to gRPC
+    verify(grpcObserver).onNext(payload);
+
+    // Verify backpressure: since gRPC was ready, it should request the next chunk
+    verify(subscription).request(1);
+  }
+
+  @Test
+  public void shouldDelegateOnNextButSuspendDemandIfNotReady() {
+    bridge.onSubscribe(subscription);
+    reset(subscription);
+
+    // Simulate gRPC internal buffer being full (not ready)
+    when(grpcObserver.isReady()).thenReturn(false);
+
+    byte[] payload = "test".getBytes();
+    ByteBuffer frame = ByteBuffer.allocate(5 + payload.length);
+    frame.put((byte) 0).putInt(payload.length).put(payload).flip();
+
+    bridge.onNext(frame);
+
+    verify(grpcObserver).onNext(payload);
+
+    // Verify backpressure: gRPC is NOT ready, so we MUST NOT request more from Jooby
+    verify(subscription, never()).request(anyLong());
+  }
+
+  @Test
+  public void shouldResumeDemandWhenGrpcBecomesReady() {
+    bridge.onSubscribe(subscription);
+    reset(subscription);
+
+    // Capture the readiness handler registered during onSubscribe
+    ArgumentCaptor<Runnable> handlerCaptor = ArgumentCaptor.forClass(Runnable.class);
+    verify(grpcObserver).setOnReadyHandler(handlerCaptor.capture());
+    Runnable onReadyHandler = handlerCaptor.getValue();
+
+    // Simulate gRPC signaling that it is now ready
+    when(grpcObserver.isReady()).thenReturn(true);
+    onReadyHandler.run();
+
+    // Verify backpressure: the handler should resume demanding data
+    verify(subscription).request(1);
+  }
+
+  @Test
+  public void shouldCancelSubscriptionAndDelegateOnError() {
+    bridge.onSubscribe(subscription);
+
+    Throwable error = new RuntimeException("Network failure");
+    bridge.onError(error);
+
+    verify(grpcObserver).onError(error);
+  }
+
+  @Test
+  public void shouldDelegateOnCompleteIdempotently() {
+    bridge.onComplete();
+    bridge.onComplete(); // Second call should be ignored
+
+    verify(grpcObserver, times(1)).onCompleted();
+  }
+}