[grid] Add BinaryMessage.wrap for transferring an owned byte array

java/src/org/openqa/selenium/netty/server/MessageInboundConverter.java

@@ -71,7 +71,8 @@ protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) {
           }
 
           if (finalFragment) {
-            message = new BinaryMessage(buffer.toByteArray());
+            // toByteArray() returns a fresh copy we own; transfer it without re-copying.
+            message = BinaryMessage.wrap(buffer.toByteArray());
             buffer.reset();
             next = Continuation.None;
           } else {

java/src/org/openqa/selenium/netty/server/WebSocketFrameProxy.java

@@ -138,7 +138,8 @@ private void forwardFrame(WebSocketFrame frame) {
             throw new UncheckedIOException("failed to read continuation frame", e);
           }
           if (cont.isFinalFragment()) {
-            upstream.send(new BinaryMessage(binaryBuffer.toByteArray()));
+            // toByteArray() returns a fresh copy we own; transfer it without re-copying.
+            upstream.send(BinaryMessage.wrap(binaryBuffer.toByteArray()));
             binaryBuffer.reset();
             next = Continuation.None;
           }

java/src/org/openqa/selenium/remote/http/BinaryMessage.java

@@ -37,6 +37,28 @@ public BinaryMessage(byte[] data) {
     System.arraycopy(data, 0, this.data, 0, data.length);
   }
 
+  /**
+   * Returns a {@link BinaryMessage} backed directly by {@code data} with no defensive copy. The
+   * caller transfers ownership of the array — once wrapped it must not be mutated, otherwise a
+   * downstream reader will see the mutation. Intended for callers that have just produced a fresh
+   * array (for example {@link java.io.ByteArrayOutputStream#toByteArray()}) and want to avoid the
+   * second allocation that the public constructor performs.
+   */
+  public static BinaryMessage wrap(byte[] data) {
+    return new BinaryMessage(Require.nonNull("Data to use", data), Ownership.TRANSFER);
+  }
+
+  // Sentinel for the private no-copy constructor below — gives it a distinct signature from
+  // the public defensive-copy BinaryMessage(byte[]) and documents the contract at the call site.
+  private enum Ownership {
+    TRANSFER
+  }
+
+  private BinaryMessage(byte[] data, Ownership ownership) {
+    Require.nonNull("Ownership", ownership);
+    this.data = data;
+  }
+
   public byte[] data() {
     return data;
   }

java/test/org/openqa/selenium/remote/http/BinaryMessageTest.java

@@ -35,4 +35,15 @@ void copiesOnlyTheReadableRegionOfABuffer() {
 
     assertThat(message.data()).containsExactly(1, 2, 3, 4);
   }
+
+  @Test
+  void wrapTransfersOwnershipWithoutCopying() {
+    byte[] payload = {9, 8, 7};
+
+    BinaryMessage wrapped = BinaryMessage.wrap(payload);
+
+    // wrap() shares storage with the caller; the constructor still defends with a copy.
+    assertThat(wrapped.data()).isSameAs(payload);
+    assertThat(new BinaryMessage(payload).data()).isNotSameAs(payload);
+  }
 }