* If you are implementing a mutation, you are obliged to handle * the incoming Callback appropriately. + *
+ * The {@link Frame}'s payload {@link ByteBuffer} is consumed by this call. * * @param frame the frame to eventually write to the network layer. * @param callback the callback to notify when the frame is written. @@ -40,8 +44,10 @@ default void sendFrame(Frame frame, Callback callback, boolean batch) /** * Send an {@link OutgoingEntry} containing a {@link Frame} and {@link Callback}. + *
+ * The {@link Frame}'s payload {@link ByteBuffer} is consumed by this call. * * @param entry the frame to eventually write to the network layer. */ void sendFrame(OutgoingEntry entry); -} +} \ No newline at end of file diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java index 2cc13cb85303..d9a6782ad019 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FragmentExtension.java @@ -94,8 +94,6 @@ public void init(ExtensionConfig config, WebSocketComponents components) public class FragmentingDemander extends WebSocketDemander { - private ByteBuffer _payload; - public FragmentingDemander() { super(FragmentExtension.this::nextIncomingFrame); @@ -112,11 +110,10 @@ protected boolean handle(Frame frame, Callback callback, boolean first) emitFrame(frame, callback); return true; } - - _payload = frame.getPayload(); } - int remaining = _payload.remaining(); + ByteBuffer payload = frame.getPayload(); + int remaining = payload.remaining(); int fragmentSize = (int)Math.min(remaining, maxFrameSize); byte opCode = (frame.getOpCode() == OpCode.CONTINUATION || !first) ? OpCode.CONTINUATION : frame.getOpCode(); Frame fragment = new Frame(opCode); @@ -126,19 +123,13 @@ protected boolean handle(Frame frame, Callback callback, boolean first) if (finished) { // If finished we don't need to fragment, forward original payload. - fragment.setPayload(_payload); - _payload = null; + fragment.setPayload(payload); } else { // Slice the fragmented payload from the buffer. - int limit = _payload.limit(); - int newLimit = _payload.position() + fragmentSize; - _payload.limit(newLimit); - ByteBuffer payloadFragment = _payload.slice(); - _payload.limit(limit); - fragment.setPayload(payloadFragment); - _payload.position(newLimit); + fragment.setPayload(payload.slice(payload.position(), fragmentSize)); + payload.position(payload.position() + fragmentSize); if (LOG.isDebugEnabled()) LOG.debug("Fragmented {}->{}", frame, fragment); } @@ -157,12 +148,5 @@ protected boolean handle(Frame frame, Callback callback, boolean first) emitFrame(fragment, payloadCallback); return finished; } - - @Override - protected void onCompleteFailure(Throwable cause) - { - super.onCompleteFailure(cause); - _payload = null; - } } } diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java index 6fda274f3ab6..ec51127f3866 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/FrameFlusher.java @@ -311,7 +311,7 @@ protected Action process() throws Throwable _releasableBuffers.add(masked); _generator.generatePayload(entry.getFrame(), payload); } - buffers.add(payload.slice()); + buffers.add(payload); } // Once we have added another buffer we cannot add to the batch buffer again. diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/Generator.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/Generator.java index 2878f24c9c2d..4118f58cfd91 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/Generator.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/Generator.java @@ -163,7 +163,7 @@ public void generatePayload(Frame frame, ByteBuffer buffer) if (frame.isMasked()) maskPayload(buffer, frame); else - buffer.put(payload.slice()); + buffer.put(payload); BufferUtil.flipToFlush(buffer, pos); } @@ -198,6 +198,7 @@ private void maskPayload(ByteBuffer buffer, Frame frame) ++maskOffset; } } + payload.position(end); } } } diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java index b4b96fe54033..42e1c2d8ceeb 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/internal/PerMessageDeflateExtension.java @@ -277,7 +277,7 @@ protected boolean onFrame(OutgoingEntry entry, boolean first) } // Provide the frames payload as input to the Deflater. - getDeflater().setInput(entry.getFrame().getPayload().slice()); + getDeflater().setInput(entry.getFrame().getPayload()); } boolean finished = deflate(entry, first); @@ -404,7 +404,7 @@ protected boolean handle(Frame frame, Callback callback, boolean first) // Provide the frames payload as input to the Inflater. _tailBytes = false; - getInflater().setInput(frame.getPayload().slice()); + getInflater().setInput(frame.getPayload()); } try diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/MessageOutputStream.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/MessageOutputStream.java index c4bf56768632..1cb7d91fe34d 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/MessageOutputStream.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/messages/MessageOutputStream.java @@ -140,17 +140,6 @@ private void flush(boolean fin) throws IOException // Any flush after the first will be a CONTINUATION frame. bytesSent += initialBufferSize; ++frameCount; - - // Buffer has been sent, but buffer should not have been consumed. - try - { - assert buffer.remaining() == initialBufferSize; - buffer.clear(); - } - catch (Throwable t) - { - t.printStackTrace(); - } } } diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/FragmentingFlusher.java b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/FragmentingFlusher.java index d69edb1d5027..c24df69a9d9d 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/FragmentingFlusher.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-common/src/main/java/org/eclipse/jetty/websocket/core/util/FragmentingFlusher.java @@ -31,7 +31,6 @@ public abstract class FragmentingFlusher extends WebSocketFlusher { private static final Logger LOG = LoggerFactory.getLogger(FragmentingFlusher.class); private final Configuration _configuration; - private ByteBuffer _payload; public FragmentingFlusher(Configuration configuration) { @@ -52,11 +51,10 @@ protected boolean onFrame(OutgoingEntry entry, boolean first) forwardFrame(entry); return true; } - - _payload = frame.getPayload().slice(); } - int remaining = _payload.remaining(); + ByteBuffer payload = frame.getPayload(); + int remaining = payload.remaining(); int fragmentSize = (int)Math.min(remaining, maxFrameSize); byte opCode = (frame.getOpCode() == OpCode.CONTINUATION || !first) ? OpCode.CONTINUATION : frame.getOpCode(); Frame fragment = new Frame(opCode); @@ -66,19 +64,13 @@ protected boolean onFrame(OutgoingEntry entry, boolean first) // If we don't need to fragment just forward with original payload. if (finished) { - fragment.setPayload(_payload); - _payload = null; + fragment.setPayload(payload); } else { // Slice the fragmented payload from the buffer. - int limit = _payload.limit(); - int newLimit = _payload.position() + fragmentSize; - _payload.limit(newLimit); - ByteBuffer payloadFragment = _payload.slice(); - _payload.limit(limit); - fragment.setPayload(payloadFragment); - _payload.position(newLimit); + fragment.setPayload(payload.slice(payload.position(), fragmentSize)); + payload.position(payload.position() + fragmentSize); if (LOG.isDebugEnabled()) LOG.debug("Fragmented {}->{}", frame, fragment); } @@ -86,11 +78,4 @@ protected boolean onFrame(OutgoingEntry entry, boolean first) forwardFrame(new OutgoingEntry.Builder(entry).frame(fragment).build()); return finished; } - - @Override - protected void onCompleteFailure(Throwable cause) - { - super.onCompleteFailure(cause); - _payload = null; - } } diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/AutoFragmentTest.java b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/AutoFragmentTest.java index cd587a435dee..6c6921196e80 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/AutoFragmentTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/AutoFragmentTest.java @@ -31,6 +31,7 @@ import org.junit.jupiter.api.Test; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.lessThanOrEqualTo; @@ -106,7 +107,7 @@ public void testOutgoingAutoFragmentToMaxFrameSize() throws Exception assertThat(frame.isFin(), is(true)); // Original frame payload should not have been changed. - assertThat(sentFrame.getPayload(), is(message)); + assertThat(sentFrame.getPayload().remaining(), is(0)); clientHandler.sendClose(); assertTrue(serverHandler.closed.await(5, TimeUnit.SECONDS)); @@ -309,9 +310,12 @@ public void testOutgoingAutoFragmentWithPermessageDeflate() throws Exception frame = clientHandler.receivedFrames.poll(1, TimeUnit.SECONDS); } + // Content of the sentPayload was fully consumed. + assertThat(sendPayload.remaining(), equalTo(0)); + // We received correct payload in 2 frames. assertThat(message, is(payload)); - assertThat(message, is(sendPayload)); + assertThat(sendPayload.remaining(), is(0)); assertThat(numFrames, is(2)); clientHandler.sendClose(); diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/FrameBufferTest.java b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/FrameBufferTest.java index 5c22ddb2fff8..4c7a131c844f 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/FrameBufferTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/FrameBufferTest.java @@ -15,7 +15,6 @@ import java.nio.ByteBuffer; import java.util.Objects; -import java.util.Random; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -72,36 +71,7 @@ public void testSingleFrame() throws Exception assertThat(frame.getOpCode(), is(OpCode.BINARY)); assertThat(frame.getPayload(), is(message)); - assertThat(sendPayload, is(message)); - - clientHandler.sendClose(); - assertTrue(clientHandler.closed.await(5, TimeUnit.SECONDS)); - assertNull(clientHandler.getError()); - } - - @Test - public void testSendSameFrameMultipleTimes() throws Exception - { - TestFrameHandler clientHandler = new TestFrameHandler(); - client.connect(clientHandler, server.getUri()).get(5, TimeUnit.SECONDS); - serverHandler.open.await(5, TimeUnit.SECONDS); - clientHandler.getCoreSession().setAutoFragment(false); - serverHandler.getCoreSession().setAutoFragment(false); - - int payloadLen = 32 * 1024; - byte[] array = new byte[payloadLen]; - new Random().nextBytes(array); - ByteBuffer message = ByteBuffer.wrap(array); - - Frame frame = new Frame(OpCode.BINARY, BufferUtil.copy(message)); - for (int i = 0; i < 200; i++) - { - clientHandler.sendFrame(frame, Callback.NOOP, false); - Frame recvFrame = Objects.requireNonNull(serverHandler.receivedFrames.poll(5, TimeUnit.SECONDS)); - assertThat(recvFrame.getOpCode(), is(OpCode.BINARY)); - assertThat(recvFrame.getPayload(), is(message)); - assertThat(frame.getPayload(), is(message)); - } + assertThat(sendPayload.remaining(), is(0)); clientHandler.sendClose(); assertTrue(clientHandler.closed.await(5, TimeUnit.SECONDS)); diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/extensions/PermessageDeflateDemandTest.java b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/extensions/PermessageDeflateDemandTest.java index d23ac8427140..172cd5cc2716 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/extensions/PermessageDeflateDemandTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/extensions/PermessageDeflateDemandTest.java @@ -99,6 +99,7 @@ public void test() throws Exception assertNotNull(coreSession); // Set max frame size to autoFragment the message into multiple frames. ByteBuffer message = randomBytes(1024); + ByteBuffer messageSlice = message.slice(); coreSession.setMaxFrameSize(64); coreSession.sendFrame(new Frame(OpCode.BINARY, message).setFin(true), Callback.NOOP, false); @@ -108,7 +109,8 @@ public void test() throws Exception assertThat(serverHandler.binaryMessages.size(), equalTo(1)); ByteBuffer recvMessage = serverHandler.binaryMessages.poll(); - assertThat(recvMessage, equalTo(message)); + assertThat(message.remaining(), equalTo(0)); + assertThat(recvMessage, equalTo(messageSlice)); } private static ByteBuffer randomBytes(int size) diff --git a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/util/MessageWriterTest.java b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/util/MessageWriterTest.java index 3d7d687a2fb8..161339b30085 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/util/MessageWriterTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-core-tests/src/test/java/org/eclipse/jetty/websocket/core/util/MessageWriterTest.java @@ -213,7 +213,9 @@ public static class FrameCapture extends CoreSession.Empty @Override public void sendFrame(OutgoingEntry entry) { - frames.offer(Frame.copy(entry.getFrame())); + Frame frame = entry.getFrame(); + frames.add(Frame.copy(frame)); + frame.getPayload().position(frame.getPayload().limit()); entry.getCallback().succeeded(); } } @@ -235,9 +237,10 @@ public void sendFrame(OutgoingEntry entry) if (frame.isFin()) { - messages.offer(activeMessage.takeCompleteString(IllegalArgumentException::new)); + messages.add(activeMessage.takeCompleteString(IllegalArgumentException::new)); activeMessage = null; } + frame.getPayload().position(frame.getPayload().limit()); entry.getCallback().succeeded(); } } diff --git a/jetty-core/jetty-websocket/jetty-websocket-jetty-common/src/test/java/org/eclipse/jetty/websocket/common/OutgoingMessageCapture.java b/jetty-core/jetty-websocket/jetty-websocket-jetty-common/src/test/java/org/eclipse/jetty/websocket/common/OutgoingMessageCapture.java index 385359bf35db..e552c7a9da5b 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-jetty-common/src/test/java/org/eclipse/jetty/websocket/common/OutgoingMessageCapture.java +++ b/jetty-core/jetty-websocket/jetty-websocket-jetty-common/src/test/java/org/eclipse/jetty/websocket/common/OutgoingMessageCapture.java @@ -70,35 +70,36 @@ public OutgoingMessageCapture() public void sendFrame(OutgoingEntry entry) { Frame frame = entry.getFrame(); + ByteBuffer payload = frame.getPayload(); switch (frame.getOpCode()) { case OpCode.CLOSE: { - CloseStatus closeStatus = new CloseStatus(frame.getPayload()); + CloseStatus closeStatus = new CloseStatus(payload); String event = String.format("CLOSE:%s:%s", CloseStatus.codeString(closeStatus.getCode()), closeStatus.getReason()); LOG.debug(event); - events.offer(event); + events.add(event); break; } case OpCode.PING: { - String event = String.format("PING:%s", dataHint(frame.getPayload())); + String event = String.format("PING:%s", dataHint(payload)); LOG.debug(event); - events.offer(event); + events.add(event); break; } case OpCode.PONG: { - String event = String.format("PONG:%s", dataHint(frame.getPayload())); + String event = String.format("PONG:%s", dataHint(payload)); LOG.debug(event); - events.offer(event); + events.add(event); break; } case OpCode.TEXT: { String event = String.format("TEXT:fin=%b:len=%d", frame.isFin(), frame.getPayloadLength()); LOG.debug(event); - events.offer(event); + events.add(event); messageSink = new StringMessageSink(this, MethodHolder.from(wholeTextHandle), true); break; } @@ -106,7 +107,7 @@ public void sendFrame(OutgoingEntry entry) { String event = String.format("BINARY:fin=%b:len=%d", frame.isFin(), frame.getPayloadLength()); LOG.debug(event); - events.offer(event); + events.add(event); messageSink = new ByteBufferMessageSink(this, MethodHolder.from(wholeBinaryHandle), true); break; } @@ -114,7 +115,7 @@ public void sendFrame(OutgoingEntry entry) { String event = String.format("CONTINUATION:fin=%b:len=%d", frame.isFin(), frame.getPayloadLength()); LOG.debug(event); - events.offer(event); + events.add(event); break; } } @@ -127,6 +128,7 @@ public void sendFrame(OutgoingEntry entry) messageSink = null; } + payload.position(payload.limit()); entry.getCallback().succeeded(); } @@ -146,6 +148,7 @@ public void onWholeText(String msg) public void onWholeBinary(ByteBuffer buf, Callback callback) { this.binaryMessages.offer(BufferUtil.copy(buf)); + buf.position(buf.limit()); callback.succeed(); } diff --git a/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/LargeDeflateTest.java b/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/LargeDeflateTest.java index d73b58027f5d..32c291a3e423 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/LargeDeflateTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/LargeDeflateTest.java @@ -81,6 +81,7 @@ public void testDeflate() throws Exception EventSocket clientSocket = new EventSocket(); Session session = _client.connect(clientSocket, upgradeRequest).get(); ByteBuffer sentMessage = largePayloads(); + ByteBuffer sentMessageSlice = sentMessage.slice(); session.sendBinary(sentMessage, Callback.NOOP); session.close(StatusCode.NORMAL, "close from test", Callback.NOOP); @@ -89,7 +90,8 @@ public void testDeflate() throws Exception assertThat(_serverSocket.closeReason, is("close from test")); ByteBuffer message = _serverSocket.binaryMessages.poll(1, TimeUnit.SECONDS); - assertThat(message, is(sentMessage)); + assertThat(message, is(sentMessageSlice)); + assertThat(sentMessage.remaining(), is(0)); } private static ByteBuffer largePayloads() diff --git a/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/listeners/WebSocketListenerTest.java b/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/listeners/WebSocketListenerTest.java index d1024217affb..74982d042f37 100644 --- a/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/listeners/WebSocketListenerTest.java +++ b/jetty-core/jetty-websocket/jetty-websocket-jetty-tests/src/test/java/org/eclipse/jetty/websocket/tests/listeners/WebSocketListenerTest.java @@ -117,9 +117,11 @@ public void testBinaryListeners(Class clazz) throws Exception // Send and receive echo on client. ByteBuffer payload = BufferUtil.toBuffer("hello world"); + ByteBuffer payloadSlice = payload.slice(); clientEndpoint.session.sendBinary(payload, Callback.NOOP); ByteBuffer echoMessage = clientEndpoint.binaryMessages.poll(5, TimeUnit.SECONDS); - assertThat(echoMessage, is(payload)); + assertThat(echoMessage, is(payloadSlice)); + assertThat(payload.remaining(), is(0)); // Close normally. clientEndpoint.session.close(StatusCode.NORMAL, "standard close", Callback.NOOP); diff --git a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/common/messages/MessageWriterTest.java b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/common/messages/MessageWriterTest.java index 830aed354f14..156ab294ab6a 100644 --- a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/common/messages/MessageWriterTest.java +++ b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/common/messages/MessageWriterTest.java @@ -151,7 +151,9 @@ public static class FrameCapture extends CoreSession.Empty @Override public void sendFrame(OutgoingEntry entry) { - frames.offer(Frame.copy(entry.getFrame())); + Frame frame = entry.getFrame(); + frames.add(Frame.copy(frame)); + frame.getPayload().position(frame.getPayload().limit()); entry.getCallback().succeeded(); } } @@ -169,7 +171,7 @@ public void sendFrame(OutgoingEntry entry) if (frame.getOpCode() == OpCode.TEXT) activeMessage = new Utf8StringBuilder(); - activeMessage.append(frame.getPayload().slice()); + activeMessage.append(frame.getPayload()); if (frame.isFin()) { diff --git a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/SingleMessageHandlerTest.java b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/SingleMessageHandlerTest.java index a75d48151f52..cf113f9b5bb5 100644 --- a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/SingleMessageHandlerTest.java +++ b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/SingleMessageHandlerTest.java @@ -100,14 +100,14 @@ public void testBinary() throws Exception // Can send/receive binary message successfully. ByteBuffer binaryMessage = BufferUtil.toBuffer("hello world"); - session.getBasicRemote().sendBinary(binaryMessage); + session.getBasicRemote().sendBinary(binaryMessage.slice()); assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessage)); // Text message is discarded by implementation. session.getBasicRemote().sendText("hello world"); // Next binary message is still received. - session.getBasicRemote().sendBinary(binaryMessage); + session.getBasicRemote().sendBinary(binaryMessage.slice()); assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessage)); session.close(); diff --git a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/coders/DecoderListTest.java b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/coders/DecoderListTest.java index 8c7a8783ed94..d5565d3bf7bf 100644 --- a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/coders/DecoderListTest.java +++ b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/coders/DecoderListTest.java @@ -52,6 +52,7 @@ import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; public class DecoderListTest { @@ -152,6 +153,8 @@ public void testBinaryDecoderList(String request, String expected) throws Except session.getBasicRemote().sendBinary(BufferUtil.toBuffer(request)); ByteBuffer response = clientEndpoint.binaryMessages.poll(3, TimeUnit.SECONDS); assertThat(BufferUtil.toString(response), is(expected)); + session.close(); + assertTrue(clientEndpoint.closeLatch.await(5, TimeUnit.SECONDS)); } @Test diff --git a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/handlers/MessageHandlerTest.java b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/handlers/MessageHandlerTest.java index b83a7f11a788..2fa876e0cb3b 100644 --- a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/handlers/MessageHandlerTest.java +++ b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee10/websocket/jakarta/tests/handlers/MessageHandlerTest.java @@ -105,9 +105,11 @@ public void testBinaryHandlers(Class clazz) throws Exception // Send and receive echo on client. ByteBuffer payload = BufferUtil.toBuffer("hello world"); + ByteBuffer payloadSlice = payload.slice(); session.getBasicRemote().sendBinary(payload); ByteBuffer echoMessage = clientEndpoint.binaryMessages.poll(5, TimeUnit.SECONDS); - assertThat(echoMessage, is(payload)); + assertThat(echoMessage, is(payloadSlice)); + assertThat(payload.remaining(), is(0)); // Close normally. session.close(new CloseReason(NORMAL_CLOSURE, "standard close")); diff --git a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee10/websocket/tests/PermessageDeflateBufferTest.java b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee10/websocket/tests/PermessageDeflateBufferTest.java index 0c07048fc5b9..da4254dfe0df 100644 --- a/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee10/websocket/tests/PermessageDeflateBufferTest.java +++ b/jetty-ee10/jetty-ee10-websocket/jetty-ee10-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee10/websocket/tests/PermessageDeflateBufferTest.java @@ -170,9 +170,11 @@ public void testPermessageDeflateFragmentedBinaryMessage() throws Exception Session session = client.connect(socket, clientUpgradeRequest).get(5, TimeUnit.SECONDS); ByteBuffer message = randomBytes(1024); + ByteBuffer messageSlice = message.slice(); session.setMaxFrameSize(64); session.sendBinary(message, Callback.NOOP); - assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(message)); + assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(messageSlice)); + assertThat(message.remaining(), equalTo(0)); session.close(); assertTrue(socket.closeLatch.await(5, TimeUnit.SECONDS)); diff --git a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/common/messages/MessageWriterTest.java b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/common/messages/MessageWriterTest.java index f6a5fb2b7e43..5baef10be4f5 100644 --- a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/common/messages/MessageWriterTest.java +++ b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/common/messages/MessageWriterTest.java @@ -151,7 +151,8 @@ public static class FrameCapture extends CoreSession.Empty @Override public void sendFrame(OutgoingEntry entry) { - frames.offer(Frame.copy(entry.getFrame())); + frames.add(Frame.copy(entry.getFrame())); + entry.getFrame().getPayload().position(entry.getFrame().getPayload().limit()); entry.getCallback().succeeded(); } } @@ -173,9 +174,10 @@ public void sendFrame(OutgoingEntry entry) if (frame.isFin()) { - messages.offer(activeMessage.takeCompleteString(() -> new BadPayloadException("Invalid UTF-8"))); + messages.add(activeMessage.takeCompleteString(() -> new BadPayloadException("Invalid UTF-8"))); activeMessage = null; } + frame.getPayload().position(frame.getPayload().limit()); entry.getCallback().succeeded(); } } diff --git a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/SingleMessageHandlerTest.java b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/SingleMessageHandlerTest.java index 5966116ab95c..c58a152625fc 100644 --- a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/SingleMessageHandlerTest.java +++ b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/SingleMessageHandlerTest.java @@ -100,9 +100,10 @@ public void testBinary() throws Exception // Can send/receive binary message successfully. ByteBuffer binaryMessage = BufferUtil.toBuffer("hello world"); + ByteBuffer binaryMessageSlice = binaryMessage.slice(); session.getBasicRemote().sendBinary(binaryMessage); - assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessage)); - + assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessageSlice)); + assertThat(binaryMessage.remaining(), equalTo(0)); // Text message is discarded by implementation. session.getBasicRemote().sendText("hello world"); diff --git a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/coders/DecoderListTest.java b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/coders/DecoderListTest.java index fb18db443491..8100f44d38c5 100644 --- a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/coders/DecoderListTest.java +++ b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/coders/DecoderListTest.java @@ -39,7 +39,6 @@ import org.eclipse.jetty.server.ServerConnector; import org.eclipse.jetty.util.BufferUtil; import org.eclipse.jetty.util.IO; -import org.eclipse.jetty.websocket.core.exception.InvalidWebSocketException; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -48,10 +47,8 @@ import org.junit.jupiter.params.provider.MethodSource; import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.containsString; -import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.is; -import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; public class DecoderListTest { @@ -152,6 +149,8 @@ public void testBinaryDecoderList(String request, String expected) throws Except session.getBasicRemote().sendBinary(BufferUtil.toBuffer(request)); ByteBuffer response = clientEndpoint.binaryMessages.poll(3, TimeUnit.SECONDS); assertThat(BufferUtil.toString(response), is(expected)); + session.close(); + assertTrue(clientEndpoint.closeLatch.await(5, TimeUnit.SECONDS)); } @Test diff --git a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/handlers/MessageHandlerTest.java b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/handlers/MessageHandlerTest.java index 6aecf2480097..b3b6ccf5d217 100644 --- a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/handlers/MessageHandlerTest.java +++ b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee11/websocket/jakarta/tests/handlers/MessageHandlerTest.java @@ -108,9 +108,11 @@ public void testBinaryHandlers(Class clazz) throws Exception // Send and receive echo on client. ByteBuffer payload = BufferUtil.toBuffer("hello world"); + ByteBuffer payloadSlice = payload.slice(); session.getBasicRemote().sendBinary(payload); ByteBuffer echoMessage = clientEndpoint.binaryMessages.poll(5, TimeUnit.SECONDS); - assertThat(echoMessage, is(payload)); + assertThat(echoMessage, is(payloadSlice)); + assertThat(payload.remaining(), is(0)); // Close normally. session.close(new CloseReason(NORMAL_CLOSURE, "standard close")); diff --git a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee11/websocket/tests/PermessageDeflateBufferTest.java b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee11/websocket/tests/PermessageDeflateBufferTest.java index 986286ce8cdb..f0030bfa01bc 100644 --- a/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee11/websocket/tests/PermessageDeflateBufferTest.java +++ b/jetty-ee11/jetty-ee11-websocket/jetty-ee11-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee11/websocket/tests/PermessageDeflateBufferTest.java @@ -170,9 +170,11 @@ public void testPermessageDeflateFragmentedBinaryMessage() throws Exception Session session = client.connect(socket, clientUpgradeRequest).get(5, TimeUnit.SECONDS); ByteBuffer message = randomBytes(1024); + ByteBuffer messageSlice = message.slice(); session.setMaxFrameSize(64); session.sendBinary(message, Callback.NOOP); - assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(message)); + assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(messageSlice)); + assertThat(message.remaining(), equalTo(0)); session.close(); assertTrue(socket.closeLatch.await(5, TimeUnit.SECONDS)); diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/common/messages/MessageWriterTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/common/messages/MessageWriterTest.java index f7ccd59c68e3..bbec95fa4e8f 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/common/messages/MessageWriterTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-common/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/common/messages/MessageWriterTest.java @@ -14,6 +14,7 @@ package org.eclipse.jetty.ee9.websocket.jakarta.common.messages; import java.io.IOException; +import java.nio.ByteBuffer; import java.util.Arrays; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -150,7 +151,9 @@ public static class FrameCapture extends CoreSession.Empty @Override public void sendFrame(OutgoingEntry entry) { - frames.offer(Frame.copy(entry.getFrame())); + frames.add(Frame.copy(entry.getFrame())); + ByteBuffer payload = entry.getFrame().getPayload(); + payload.position(payload.limit()); entry.getCallback().succeeded(); } } @@ -175,6 +178,8 @@ public void sendFrame(OutgoingEntry entry) messages.offer(activeMessage.takeCompleteString(IllegalArgumentException::new)); activeMessage = null; } + + frame.getPayload().position(frame.getPayload().limit()); entry.getCallback().succeeded(); } } diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/SingleMessageHandlerTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/SingleMessageHandlerTest.java index 957e5b72f89b..c0430e6cb878 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/SingleMessageHandlerTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/SingleMessageHandlerTest.java @@ -100,8 +100,10 @@ public void testBinary() throws Exception // Can send/receive binary message successfully. ByteBuffer binaryMessage = BufferUtil.toBuffer("hello world"); + ByteBuffer binaryMessageSlice = binaryMessage.slice(); session.getBasicRemote().sendBinary(binaryMessage); - assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessage)); + assertThat(BINARY_MESSAGES.poll(5, TimeUnit.SECONDS), equalTo(binaryMessageSlice)); + assertThat(binaryMessage.remaining(), equalTo(0)); // Text message is discarded by implementation. session.getBasicRemote().sendText("hello world"); diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/handlers/MessageHandlerTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/handlers/MessageHandlerTest.java index b4127920bee9..2e570291cb0f 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/handlers/MessageHandlerTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jakarta-tests/src/test/java/org/eclipse/jetty/ee9/websocket/jakarta/tests/handlers/MessageHandlerTest.java @@ -105,9 +105,11 @@ public void testBinaryHandlers(Class clazz) throws Exception // Send and receive echo on client. ByteBuffer payload = BufferUtil.toBuffer("hello world"); + ByteBuffer payloadSlice = payload.slice(); session.getBasicRemote().sendBinary(payload); ByteBuffer echoMessage = clientEndpoint.binaryMessages.poll(5, TimeUnit.SECONDS); - assertThat(echoMessage, is(payload)); + assertThat(echoMessage, is(payloadSlice)); + assertThat(payload.remaining(), is(0)); // Close normally. session.close(new CloseReason(NORMAL_CLOSURE, "standard close")); diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-common/src/test/java/org/eclipse/jetty/ee9/websocket/common/OutgoingMessageCapture.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-common/src/test/java/org/eclipse/jetty/ee9/websocket/common/OutgoingMessageCapture.java index aef970a84150..9647a241324d 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-common/src/test/java/org/eclipse/jetty/ee9/websocket/common/OutgoingMessageCapture.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-common/src/test/java/org/eclipse/jetty/ee9/websocket/common/OutgoingMessageCapture.java @@ -128,6 +128,7 @@ public void sendFrame(OutgoingEntry entry) messageSink = null; } + frame.getPayload().position(frame.getPayload().limit()); callback.succeeded(); } diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/LargeDeflateTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/LargeDeflateTest.java index 35d4dea52aff..4dda4c261576 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/LargeDeflateTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/LargeDeflateTest.java @@ -82,6 +82,7 @@ public void testDeflate() throws Exception EventSocket clientSocket = new EventSocket(); Session session = _client.connect(clientSocket, URI.create("ws://localhost:" + _connector.getLocalPort() + "/ws"), upgradeRequest).get(); ByteBuffer sentMessage = largePayloads(); + ByteBuffer sentMessageSlice = sentMessage.slice(); session.getRemote().sendBytes(sentMessage); session.close(StatusCode.NORMAL, "close from test"); @@ -90,7 +91,8 @@ public void testDeflate() throws Exception assertThat(_serverSocket.closeReason, is("close from test")); ByteBuffer message = _serverSocket.binaryMessages.poll(1, TimeUnit.SECONDS); - assertThat(message, is(sentMessage)); + assertThat(message, is(sentMessageSlice)); + assertThat(sentMessage.remaining(), is(0)); } @Test diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/PermessageDeflateBufferTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/PermessageDeflateBufferTest.java index 5ae517832d32..c98e71d46b00 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/PermessageDeflateBufferTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/PermessageDeflateBufferTest.java @@ -149,9 +149,11 @@ public void testPermessageDeflateFragmentedBinaryMessage() throws Exception Session session = client.connect(socket, uri, clientUpgradeRequest).get(5, TimeUnit.SECONDS); ByteBuffer message = randomBytes(1024); + ByteBuffer messageSlice = message.slice(); session.setMaxFrameSize(64); session.getRemote().sendBytes(message); - assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(message)); + assertThat(socket.binaryMessages.poll(5, TimeUnit.SECONDS), equalTo(messageSlice)); + assertThat(message.remaining(), equalTo(0)); session.close(); assertTrue(socket.closeLatch.await(5, TimeUnit.SECONDS)); diff --git a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/listeners/WebSocketListenerTest.java b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/listeners/WebSocketListenerTest.java index 8fe1c551c8fa..d29c2f8de22e 100644 --- a/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/listeners/WebSocketListenerTest.java +++ b/jetty-ee9/jetty-ee9-websocket/jetty-ee9-websocket-jetty-tests/src/test/java/org/eclipse/jetty/ee9/websocket/tests/listeners/WebSocketListenerTest.java @@ -118,9 +118,11 @@ public void testBinaryListeners(Class clazz) throws Exception // Send and receive echo on client. ByteBuffer payload = BufferUtil.toBuffer("hello world"); + ByteBuffer payloadSlice = payload.slice(); clientEndpoint.session.getRemote().sendBytes(payload); ByteBuffer echoMessage = clientEndpoint.binaryMessages.poll(5, TimeUnit.SECONDS); - assertThat(echoMessage, is(payload)); + assertThat(echoMessage, is(payloadSlice)); + assertThat(payload.remaining(), is(0)); // Close normally. clientEndpoint.session.close(StatusCode.NORMAL, "standard close");