diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java b/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java index 01ed68fd..4174bbe2 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/RTCDataChannel.java +++ b/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); } } diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelBufferWindowTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/RTCDataChannelBufferWindowTests.java new file mode 100644 index 00000000..b43786cf --- /dev/null +++ b/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); + } +} diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java b/webrtc/src/test/java/dev/onvoid/webrtc/TestDataChannelPair.java new file mode 100644 index 00000000..c5afa1e0 --- /dev/null +++ b/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 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(); + } +}