perf: avoid redundant data channel buffer copies

webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java

@@ -145,8 +145,7 @@ private RTCDataChannel() {
 	/**
 	 * Sends data in the provided buffer to the remote peer. Only the bytes
 	 * between the buffer's position and limit are sent, for heap and direct
-	 * buffers alike. The buffer is read through a duplicate, so the caller's
-	 * position is left untouched.
+	 * buffers alike. The caller's position is left untouched.
 	 *
 	 * @param buffer The buffer to be queued for transmission.
 	 *
@@ -161,10 +160,7 @@ public void send(RTCDataChannelBuffer buffer) throws Exception {
 				sendDirectBuffer(data, buffer.binary);
 			}
 			else {
-				ByteBuffer window = ByteBuffer.allocateDirect(data.remaining());
-				window.put(data.duplicate());
-				window.flip();
-				sendDirectBuffer(window, buffer.binary);
+				sendDirectBuffer(data.slice(), buffer.binary);
 			}
 		}
 		else {
@@ -215,12 +211,7 @@ public void sendAsync(RTCDataChannelBuffer buffer) {
 			sendDirectBufferAsync(data, data.position(), data.remaining(), buffer.binary);
 		}
 		else {
-			// The byte array path transmits whole arrays, so copy exactly
-			// the readable window, position to limit; a duplicate leaves
-			// the caller's position untouched.
-			byte[] window = new byte[data.remaining()];
-			data.duplicate().get(window);
-			sendByteArrayBufferAsync(window, buffer.binary);
+			sendByteArrayBufferAsync(copyWindow(data), buffer.binary);
 		}
 	}
 

webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelBufferWindowTests.java

@@ -0,0 +1,79 @@
+package dev.onvoid.webrtc;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.Test;
+
+class RTCDataChannelBufferWindowTests extends TestBase {
+
+	@Test
+	void synchronousBufferWindows() throws Exception {
+		assertBufferWindows(false);
+	}
+
+	@Test
+	void asynchronousBufferWindows() throws Exception {
+		assertBufferWindows(true);
+	}
+
+	private void assertBufferWindows(boolean asynchronous) throws Exception {
+		try (TestDataChannelPair pair = new TestDataChannelPair(factory)) {
+			pair.connect();
+			for (boolean direct : new boolean[] { false, true }) {
+				for (int layout = 0; layout < 5; layout++) {
+					ByteBuffer storage = direct ? ByteBuffer.allocateDirect(16) : ByteBuffer.allocate(16);
+					for (int i = 0; i < storage.capacity(); i++) {
+						storage.put(i, (byte) (i + 1));
+					}
+					ByteBuffer data = storage.duplicate();
+					if (layout != 0) {
+						data.position(3);
+						data.limit(layout == 4 ? 3 : 12);
+					}
+					if (layout == 2) {
+						data = data.slice();
+					}
+					if (layout == 3) {
+						data = data.asReadOnlyBuffer();
+					}
+					assertWindow(pair, data, storage, asynchronous);
+				}
+				ByteBuffer empty = direct ? ByteBuffer.allocateDirect(0) : ByteBuffer.allocate(0);
+				assertWindow(pair, empty, empty, asynchronous);
+			}
+		}
+	}
+
+	private static void assertWindow(TestDataChannelPair pair, ByteBuffer data,
+			ByteBuffer storage, boolean asynchronous) throws Exception {
+		int position = data.position();
+		int limit = data.limit();
+		byte[] expected = new byte[data.remaining()];
+		data.duplicate().get(expected);
+		RTCDataChannelBuffer buffer = new RTCDataChannelBuffer(data, true);
+		if (asynchronous) {
+			pair.sender.sendAsync(buffer);
+		}
+		else {
+			pair.sender.send(buffer);
+		}
+		assertEquals(position, data.position());
+		assertEquals(limit, data.limit());
+		// The native send must own its bytes before the caller reuses the buffer.
+		for (int i = 0; i < storage.capacity(); i++) {
+			storage.put(i, (byte) 0);
+		}
+		RTCDataChannelBuffer received = pair.messages.poll(5, TimeUnit.SECONDS);
+		assertNotNull(received, "Message was not received");
+		assertTrue(received.binary);
+		byte[] actual = new byte[received.data.remaining()];
+		received.data.get(actual);
+		assertArrayEquals(expected, actual);
+	}
+}

webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java

@@ -0,0 +1,96 @@
+package dev.onvoid.webrtc;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+class TestDataChannelPair implements AutoCloseable {
+
+	final RTCDataChannel sender;
+	final RTCDataChannel receiver;
+	final BlockingQueue<RTCDataChannelBuffer> messages = new LinkedBlockingQueue<>();
+
+	private final RTCPeerConnection caller;
+	private final RTCPeerConnection callee;
+	private final CountDownLatch open = new CountDownLatch(2);
+
+	TestDataChannelPair(PeerConnectionFactory factory) {
+		RTCPeerConnection[] peers = new RTCPeerConnection[2];
+		caller = factory.createPeerConnection(new RTCConfiguration(),
+				candidate -> peers[1].addIceCandidate(candidate));
+		callee = factory.createPeerConnection(new RTCConfiguration(),
+				candidate -> peers[0].addIceCandidate(candidate));
+		peers[0] = caller;
+		peers[1] = callee;
+
+		RTCDataChannelInit config = new RTCDataChannelInit();
+		config.negotiated = true;
+		config.id = 0;
+		sender = caller.createDataChannel("send", config);
+		receiver = callee.createDataChannel("send", config);
+		sender.registerObserver(observer(sender));
+		receiver.registerObserver(observer(receiver));
+	}
+
+	void connect() throws Exception {
+		TestCreateDescObserver offer = new TestCreateDescObserver();
+		caller.createOffer(new RTCOfferOptions(), offer);
+		RTCSessionDescription offerDescription = offer.get(5, TimeUnit.SECONDS);
+		setDescription(caller, offerDescription, true);
+		setDescription(callee, offerDescription, false);
+
+		TestCreateDescObserver answer = new TestCreateDescObserver();
+		callee.createAnswer(new RTCAnswerOptions(), answer);
+		RTCSessionDescription answerDescription = answer.get(5, TimeUnit.SECONDS);
+		setDescription(callee, answerDescription, true);
+		setDescription(caller, answerDescription, false);
+		assertTrue(open.await(5, TimeUnit.SECONDS), "Data channels did not open");
+	}
+
+	private static void setDescription(RTCPeerConnection peer,
+			RTCSessionDescription description, boolean local) throws Exception {
+		TestSetDescObserver observer = new TestSetDescObserver();
+		if (local) {
+			peer.setLocalDescription(description, observer);
+		}
+		else {
+			peer.setRemoteDescription(description, observer);
+		}
+		observer.get(5, TimeUnit.SECONDS);
+	}
+
+	private RTCDataChannelObserver observer(RTCDataChannel channel) {
+		return new RTCDataChannelObserver() {
+			@Override
+			public void onBufferedAmountChange(long sentDataSize) { }
+
+			@Override
+			public void onStateChange() {
+				if (channel.getState() == RTCDataChannelState.OPEN) {
+					open.countDown();
+				}
+			}
+
+			@Override
+			public void onMessage(RTCDataChannelBuffer buffer) {
+				byte[] bytes = new byte[buffer.data.remaining()];
+				buffer.data.duplicate().get(bytes);
+				messages.add(new RTCDataChannelBuffer(ByteBuffer.wrap(bytes), buffer.binary));
+			}
+		};
+	}
+
+	@Override
+	public void close() {
+		sender.unregisterObserver();
+		receiver.unregisterObserver();
+		caller.close();
+		callee.close();
+		sender.dispose();
+		receiver.dispose();
+	}
+}