Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4593 from eclipse/jetty-10.0.x-4571-MessageSink 

Issue #4571 - websocket aggregating text and binary MessageSinks
This commit is contained in:
Lachlan 2020-03-11 14:14:15 +11:00 committed by GitHub
parent 360b878d02 adbb3f165e
commit b0ddba49da
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 403 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/Frame.java
View File

@ -221,6 +221,12 @@ public class Frame
return payload;
}
/**
* Get the payload of the frame as a UTF-8 string.
* <p>Should only be used in testing, does not validate the
* UTF-8 and a non fin frame can contain partial UTF-8 characters.</p>
* @return the payload as a UTF-8 string.
*/
public String getPayloadAsUTF8()
{
if (payload == null)

2
jetty-websocket/websocket-javax-tests/src/test/java/org/eclipse/jetty/websocket/javax/tests/client/MessageReceivingTest.java
View File

@ -477,7 +477,7 @@ public class MessageReceivingTest
@Override
public void onMessage(ByteBuffer message)
{
final String stringResult = new String(message.array());
final String stringResult = BufferUtil.toString(message);
messageQueue.offer(stringResult);
}
}

48
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/ByteArrayMessageSink.java
View File

@ -19,6 +19,7 @@
package org.eclipse.jetty.websocket.util.messages;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodType;
import java.nio.ByteBuffer;
@ -49,40 +50,40 @@ public class ByteArrayMessageSink extends AbstractMessageSink
}
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
if (frame.hasPayload())
size += frame.getPayloadLength();
long maxBinaryMessageSize = session.getMaxBinaryMessageSize();
if (maxBinaryMessageSize > 0 && size > maxBinaryMessageSize)
{
ByteBuffer payload = frame.getPayload();
size += payload.remaining();
long maxBinaryMessageSize = session.getMaxBinaryMessageSize();
if (maxBinaryMessageSize > 0 && size > maxBinaryMessageSize)
{
throw new MessageTooLargeException(String.format("Binary message too large: (actual) %,d > (configured max binary buffer size) %,d",
size, maxBinaryMessageSize));
}
if (out == null)
out = new ByteArrayOutputStream(BUFFER_SIZE);
BufferUtil.writeTo(payload, out);
throw new MessageTooLargeException(String.format("Binary message too large: (actual) %,d > (configured max binary message size) %,d",
size, maxBinaryMessageSize));
}
if (frame.isFin())
// If we are fin and no OutputStream has been created we don't need to aggregate.
if (frame.isFin() && (out == null))
{
if (out != null)
if (frame.hasPayload())
{
byte[] buf = out.toByteArray();
byte[] buf = BufferUtil.toArray(frame.getPayload());
methodHandle.invoke(buf, 0, buf.length);
}
else
methodHandle.invoke(EMPTY_BUFFER, 0, 0);
callback.succeeded();
return;
}
aggregatePayload(frame);
if (frame.isFin())
{
byte[] buf = out.toByteArray();
methodHandle.invoke(buf, 0, buf.length);
}
callback.succeeded();
}
catch (Throwable t)
@ -99,4 +100,15 @@ public class ByteArrayMessageSink extends AbstractMessageSink
}
}
}
private void aggregatePayload(Frame frame) throws IOException
{
if (frame.hasPayload())
{
ByteBuffer payload = frame.getPayload();
if (out == null)
out = new ByteArrayOutputStream(BUFFER_SIZE);
BufferUtil.writeTo(payload, out);
}
}
}

50
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/ByteBufferMessageSink.java
View File

@ -19,6 +19,7 @@
package org.eclipse.jetty.websocket.util.messages;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodType;
import java.nio.ByteBuffer;
@ -50,38 +51,35 @@ public class ByteBufferMessageSink extends AbstractMessageSink
}
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
if (frame.hasPayload())
size += frame.getPayloadLength();
long maxBinaryMessageSize = session.getMaxBinaryMessageSize();
if (maxBinaryMessageSize > 0 && size > maxBinaryMessageSize)
{
ByteBuffer payload = frame.getPayload();
size += payload.remaining();
long maxBinaryMessageSize = session.getMaxBinaryMessageSize();
if (maxBinaryMessageSize > 0 && size > maxBinaryMessageSize)
{
throw new MessageTooLargeException(String.format("Binary message too large: (actual) %,d > (configured max binary message size) %,d",
size, maxBinaryMessageSize));
}
if (out == null)
out = new ByteArrayOutputStream(BUFFER_SIZE);
BufferUtil.writeTo(payload, out);
payload.position(payload.limit()); // consume buffer
throw new MessageTooLargeException(String.format("Binary message too large: (actual) %,d > (configured max binary message size) %,d",
size, maxBinaryMessageSize));
}
if (frame.isFin())
// If we are fin and no OutputStream has been created we don't need to aggregate.
if (frame.isFin() && (out == null))
{
if (out != null)
methodHandle.invoke(ByteBuffer.wrap(out.toByteArray()));
if (frame.hasPayload())
methodHandle.invoke(frame.getPayload());
else
methodHandle.invoke(BufferUtil.EMPTY_BUFFER);
callback.succeeded();
return;
}
aggregatePayload(frame);
if (frame.isFin())
methodHandle.invoke(ByteBuffer.wrap(out.toByteArray()));
callback.succeeded();
}
catch (Throwable t)
@ -98,4 +96,18 @@ public class ByteBufferMessageSink extends AbstractMessageSink
}
}
}
private void aggregatePayload(Frame frame) throws IOException
{
if (frame.hasPayload())
{
ByteBuffer payload = frame.getPayload();
if (out == null)
out = new ByteArrayOutputStream(BUFFER_SIZE);
BufferUtil.writeTo(payload, out);
payload.position(payload.limit()); // consume buffer
}
}
}

15
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/DispatchedMessageSink.java
View File

@ -136,20 +136,7 @@ public abstract class DispatchedMessageSink<T> extends AbstractMessageSink
if (frame.isFin())
{
CompletableFuture<Void> finComplete = new CompletableFuture<>();
frameCallback = new Callback()
{
@Override
public void failed(Throwable cause)
{
finComplete.completeExceptionally(cause);
}
@Override
public void succeeded()
{
finComplete.complete(null);
}
};
frameCallback = Callback.from(() -> finComplete.complete(null), finComplete::completeExceptionally);
CompletableFuture.allOf(dispatchComplete, finComplete).whenComplete(
(aVoid, throwable) ->
{

20
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/PartialByteArrayMessageSink.java
View File

@ -20,7 +20,6 @@ package org.eclipse.jetty.websocket.util.messages;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodType;
import java.nio.ByteBuffer;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
@ -30,6 +29,8 @@ import org.eclipse.jetty.websocket.util.InvalidSignatureException;
public class PartialByteArrayMessageSink extends AbstractMessageSink
{
private static byte[] EMPTY_BUFFER = new byte[0];
public PartialByteArrayMessageSink(CoreSession session, MethodHandle methodHandle)
{
super(session, methodHandle);
@ -42,28 +43,17 @@ public class PartialByteArrayMessageSink extends AbstractMessageSink
}
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
byte[] buffer;
int offset = 0;
int length = 0;
if (frame.hasPayload())
if (frame.hasPayload() || frame.isFin())
{
ByteBuffer payload = frame.getPayload();
length = payload.remaining();
buffer = BufferUtil.toArray(payload);
}
else
{
buffer = new byte[0];
byte[] buffer = frame.hasPayload() ? BufferUtil.toArray(frame.getPayload()) : EMPTY_BUFFER;
methodHandle.invoke(buffer, 0, buffer.length, frame.isFin());
}
methodHandle.invoke(buffer, offset, length, frame.isFin());
callback.succeeded();
}
catch (Throwable t)

22
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/PartialByteBufferMessageSink.java
View File

@ -19,9 +19,7 @@
package org.eclipse.jetty.websocket.util.messages;
import java.lang.invoke.MethodHandle;
import java.nio.ByteBuffer;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.websocket.core.CoreSession;
import org.eclipse.jetty.websocket.core.Frame;
@ -41,29 +39,13 @@ public class PartialByteBufferMessageSink extends AbstractMessageSink
*/
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
ByteBuffer buffer;
if (frame.hasPayload())
{
ByteBuffer payload = frame.getPayload();
// copy buffer here
buffer = ByteBuffer.allocate(payload.remaining());
BufferUtil.clearToFill(buffer);
BufferUtil.put(payload, buffer);
BufferUtil.flipToFlush(buffer, 0);
}
else
{
buffer = BufferUtil.EMPTY_BUFFER;
}
methodHandle.invoke(buffer, frame.isFin());
if (frame.hasPayload() || frame.isFin())
methodHandle.invoke(frame.getPayload(), frame.isFin());
callback.succeeded();
}

47
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/PartialStringMessageSink.java
View File

@ -19,73 +19,40 @@
package org.eclipse.jetty.websocket.util.messages;
import java.lang.invoke.MethodHandle;
import java.nio.ByteBuffer;
import java.util.Objects;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.core.CoreSession;
import org.eclipse.jetty.websocket.core.Frame;
public class PartialStringMessageSink extends AbstractMessageSink
{
private static final Logger LOG = Log.getLogger(PartialStringMessageSink.class);
private Utf8StringBuilder utf;
private int size;
private Utf8StringBuilder out;
public PartialStringMessageSink(CoreSession session, MethodHandle methodHandle)
{
super(session, methodHandle);
Objects.requireNonNull(methodHandle, "MethodHandle");
this.size = 0;
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
if (utf == null)
utf = new Utf8StringBuilder(1024);
if (frame.hasPayload())
{
ByteBuffer payload = frame.getPayload();
//TODO we should fragment on maxTextMessageBufferSize not limit
//TODO also for PartialBinaryMessageSink
/*
if ((session.getMaxTextMessageBufferSize() > 0) && (size + payload.remaining() > session.getMaxTextMessageBufferSize()))
{
throw new MessageTooLargeException(String.format("Binary message too large: (actual) %,d > (configured max text buffer size) %,d",
size + payload.remaining(), session.getMaxTextMessageBufferSize()));
}
*/
size += payload.remaining();
if (LOG.isDebugEnabled())
LOG.debug("Raw Payload {}", BufferUtil.toDetailString(payload));
// allow for fast fail of BAD utf
utf.append(payload);
}
if (out == null)
out = new Utf8StringBuilder(session.getInputBufferSize());
out.append(frame.getPayload());
if (frame.isFin())
{
// Using toString to trigger failure on incomplete UTF-8
methodHandle.invoke(utf.toString(), true);
// reset
size = 0;
utf = null;
methodHandle.invoke(out.toString(), true);
out = null;
}
else
{
methodHandle.invoke(utf.takePartialString(), false);
methodHandle.invoke(out.takePartialString(), false);
}
callback.succeeded();

57
jetty-websocket/websocket-util/src/main/java/org/eclipse/jetty/websocket/util/messages/StringMessageSink.java
View File

@ -19,21 +19,16 @@
package org.eclipse.jetty.websocket.util.messages;
import java.lang.invoke.MethodHandle;
import java.nio.ByteBuffer;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.Utf8StringBuilder;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.core.CoreSession;
import org.eclipse.jetty.websocket.core.Frame;
import org.eclipse.jetty.websocket.core.exception.MessageTooLargeException;
public class StringMessageSink extends AbstractMessageSink
{
private static final Logger LOG = Log.getLogger(StringMessageSink.class);
private Utf8StringBuilder utf;
private Utf8StringBuilder out;
private int size;
public StringMessageSink(CoreSession session, MethodHandle methodHandle)
@ -42,46 +37,25 @@ public class StringMessageSink extends AbstractMessageSink
this.size = 0;
}
@SuppressWarnings("Duplicates")
@Override
public void accept(Frame frame, Callback callback)
{
try
{
if (frame.hasPayload())
size += frame.getPayloadLength();
long maxTextMessageSize = session.getMaxTextMessageSize();
if (maxTextMessageSize > 0 && size > maxTextMessageSize)
{
ByteBuffer payload = frame.getPayload();
size += payload.remaining();
long maxTextMessageSize = session.getMaxTextMessageSize();
if (maxTextMessageSize > 0 && size > maxTextMessageSize)
{
throw new MessageTooLargeException(String.format("Text message too large: (actual) %,d > (configured max text message size) %,d",
size, maxTextMessageSize));
}
if (utf == null)
utf = new Utf8StringBuilder(1024);
if (LOG.isDebugEnabled())
LOG.debug("Raw Payload {}", BufferUtil.toDetailString(payload));
// allow for fast fail of BAD utf (incomplete utf will trigger on messageComplete)
utf.append(payload);
throw new MessageTooLargeException(String.format("Text message too large: (actual) %,d > (configured max text message size) %,d",
size, maxTextMessageSize));
}
if (out == null)
out = new Utf8StringBuilder(session.getInputBufferSize());
out.append(frame.getPayload());
if (frame.isFin())
{
// notify event
if (utf != null)
methodHandle.invoke(utf.toString());
else
methodHandle.invoke("");
// reset
size = 0;
utf = null;
}
methodHandle.invoke(out.toString());
callback.succeeded();
}
@ -89,5 +63,14 @@ public class StringMessageSink extends AbstractMessageSink
{
callback.failed(t);
}
finally
{
if (frame.isFin())
{
// reset
size = 0;
out = null;
}
}
}
}

153
jetty-websocket/websocket-util/src/test/org/eclipse/jetty/websocket/util/PartialStringMessageSinkTest.java Normal file
View File

@ -0,0 +1,153 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.websocket.util;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.FutureCallback;
import org.eclipse.jetty.util.Utf8Appendable;
import org.eclipse.jetty.websocket.core.CoreSession;
import org.eclipse.jetty.websocket.core.Frame;
import org.eclipse.jetty.websocket.core.OpCode;
import org.eclipse.jetty.websocket.util.messages.PartialStringMessageSink;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
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 PartialStringMessageSinkTest
{
private CoreSession coreSession = new CoreSession.Empty();
private OnMessageEndpoint endpoint = new OnMessageEndpoint();
private PartialStringMessageSink messageSink;
@BeforeEach
public void before() throws Exception
{
messageSink = new PartialStringMessageSink(coreSession, endpoint.getMethodHandle());
}
@Test
public void testValidUtf8() throws Exception
{
ByteBuffer utf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90, (byte)0x8D, (byte)0x88});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, utf8Payload).setFin(true), callback);
callback.block(5, TimeUnit.SECONDS);
List<String> message = Objects.requireNonNull(endpoint.messages.poll(5, TimeUnit.SECONDS));
assertThat(message.size(), is(1));
assertThat(message.get(0), is("\uD800\uDF48"));
}
@Test
public void testUtf8Continuation() throws Exception
{
ByteBuffer firstUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90});
ByteBuffer continuationUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0x8D, (byte)0x88});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, firstUtf8Payload).setFin(false), callback);
callback.block(5, TimeUnit.SECONDS);
callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, continuationUtf8Payload).setFin(true), callback);
callback.block(5, TimeUnit.SECONDS);
List<String> message = Objects.requireNonNull(endpoint.messages.poll(5, TimeUnit.SECONDS));
assertThat(message.size(), is(2));
assertThat(message.get(0), is(""));
assertThat(message.get(1), is("\uD800\uDF48"));
}
@Test
public void testInvalidSingleFrameUtf8() throws Exception
{
ByteBuffer invalidUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90, (byte)0x8D});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, invalidUtf8Payload).setFin(true), callback);
// Callback should fail and we don't receive the message in the sink.
RuntimeException error = assertThrows(RuntimeException.class, () -> callback.block(5, TimeUnit.SECONDS));
assertThat(error.getCause(), instanceOf(Utf8Appendable.NotUtf8Exception.class));
List<String> message = Objects.requireNonNull(endpoint.messages.poll(5, TimeUnit.SECONDS));
assertTrue(message.isEmpty());
}
@Test
public void testInvalidMultiFrameUtf8() throws Exception
{
ByteBuffer firstUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90});
ByteBuffer continuationUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0x8D});
FutureCallback firstCallback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, firstUtf8Payload).setFin(false), firstCallback);
firstCallback.block(5, TimeUnit.SECONDS);
FutureCallback continuationCallback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, continuationUtf8Payload).setFin(true), continuationCallback);
// Callback should fail and we only received the first frame which had no full character.
RuntimeException error = assertThrows(RuntimeException.class, () -> continuationCallback.block(5, TimeUnit.SECONDS));
assertThat(error.getCause(), instanceOf(Utf8Appendable.NotUtf8Exception.class));
List<String> message = Objects.requireNonNull(endpoint.messages.poll(5, TimeUnit.SECONDS));
assertThat(message.size(), is(1));
assertThat(message.get(0), is(""));
}
public static class OnMessageEndpoint
{
private BlockingArrayQueue<List<String>> messages;
public OnMessageEndpoint()
{
messages = new BlockingArrayQueue<>();
messages.add(new ArrayList<>());
}
public void onMessage(String message, boolean last)
{
messages.get(messages.size() - 1).add(message);
if (last)
messages.add(new ArrayList<>());
}
public MethodHandle getMethodHandle() throws Exception
{
return MethodHandles.lookup()
.findVirtual(this.getClass(), "onMessage", MethodType.methodType(void.class, String.class, boolean.class))
.bindTo(this);
}
}
}

147
jetty-websocket/websocket-util/src/test/org/eclipse/jetty/websocket/util/StringMessageSinkTest.java Normal file
View File

@ -0,0 +1,147 @@
//
// ========================================================================
// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
//
// This program and the accompanying materials are made available under
// the terms of the Eclipse Public License 2.0 which is available at
// https://www.eclipse.org/legal/epl-2.0
//
// This Source Code may also be made available under the following
// Secondary Licenses when the conditions for such availability set
// forth in the Eclipse Public License, v. 2.0 are satisfied:
// the Apache License v2.0 which is available at
// https://www.apache.org/licenses/LICENSE-2.0
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
// ========================================================================
//
package org.eclipse.jetty.websocket.util;
import java.lang.invoke.MethodHandle;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodType;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import org.eclipse.jetty.util.BlockingArrayQueue;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.FutureCallback;
import org.eclipse.jetty.util.Utf8Appendable;
import org.eclipse.jetty.websocket.core.CoreSession;
import org.eclipse.jetty.websocket.core.Frame;
import org.eclipse.jetty.websocket.core.OpCode;
import org.eclipse.jetty.websocket.core.exception.MessageTooLargeException;
import org.eclipse.jetty.websocket.util.messages.StringMessageSink;
import org.junit.jupiter.api.Test;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class StringMessageSinkTest
{
private CoreSession coreSession = new CoreSession.Empty();
private OnMessageEndpoint endpoint = new OnMessageEndpoint();
@Test
public void testMaxMessageSize() throws Exception
{
StringMessageSink messageSink = new StringMessageSink(coreSession, endpoint.getMethodHandle());
ByteBuffer utf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90, (byte)0x8D, (byte)0x88});
FutureCallback callback = new FutureCallback();
coreSession.setMaxTextMessageSize(3);
messageSink.accept(new Frame(OpCode.TEXT, utf8Payload).setFin(true), callback);
// Callback should fail and we don't receive the message in the sink.
RuntimeException error = assertThrows(RuntimeException.class, () -> callback.block(5, TimeUnit.SECONDS));
assertThat(error.getCause(), instanceOf(MessageTooLargeException.class));
assertNull(endpoint.messages.poll());
}
@Test
public void testValidUtf8() throws Exception
{
StringMessageSink messageSink = new StringMessageSink(coreSession, endpoint.getMethodHandle());
ByteBuffer utf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90, (byte)0x8D, (byte)0x88});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, utf8Payload).setFin(true), callback);
callback.block(5, TimeUnit.SECONDS);
assertThat(endpoint.messages.poll(5, TimeUnit.SECONDS), is("\uD800\uDF48"));
}
@Test
public void testUtf8Continuation() throws Exception
{
StringMessageSink messageSink = new StringMessageSink(coreSession, endpoint.getMethodHandle());
ByteBuffer firstUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90});
ByteBuffer continuationUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0x8D, (byte)0x88});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, firstUtf8Payload).setFin(false), callback);
callback.block(5, TimeUnit.SECONDS);
callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, continuationUtf8Payload).setFin(true), callback);
callback.block(5, TimeUnit.SECONDS);
assertThat(endpoint.messages.poll(5, TimeUnit.SECONDS), is("\uD800\uDF48"));
}
@Test
public void testInvalidSingleFrameUtf8() throws Exception
{
StringMessageSink messageSink = new StringMessageSink(coreSession, endpoint.getMethodHandle());
ByteBuffer invalidUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90, (byte)0x8D});
FutureCallback callback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, invalidUtf8Payload).setFin(true), callback);
// Callback should fail and we don't receive the message in the sink.
RuntimeException error = assertThrows(RuntimeException.class, () -> callback.block(5, TimeUnit.SECONDS));
assertThat(error.getCause(), instanceOf(Utf8Appendable.NotUtf8Exception.class));
assertNull(endpoint.messages.poll());
}
@Test
public void testInvalidMultiFrameUtf8() throws Exception
{
StringMessageSink messageSink = new StringMessageSink(coreSession, endpoint.getMethodHandle());
ByteBuffer firstUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0xF0, (byte)0x90});
ByteBuffer continuationUtf8Payload = BufferUtil.toBuffer(new byte[]{(byte)0x8D});
FutureCallback firstCallback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, firstUtf8Payload).setFin(false), firstCallback);
firstCallback.block(5, TimeUnit.SECONDS);
FutureCallback continuationCallback = new FutureCallback();
messageSink.accept(new Frame(OpCode.TEXT, continuationUtf8Payload).setFin(true), continuationCallback);
// Callback should fail and we don't receive the message in the sink.
RuntimeException error = assertThrows(RuntimeException.class, () -> continuationCallback.block(5, TimeUnit.SECONDS));
assertThat(error.getCause(), instanceOf(Utf8Appendable.NotUtf8Exception.class));
assertNull(endpoint.messages.poll());
}
public static class OnMessageEndpoint
{
private BlockingArrayQueue<String> messages = new BlockingArrayQueue<>();
public void onMessage(String message)
{
messages.add(message);
}
public MethodHandle getMethodHandle() throws Exception
{
return MethodHandles.lookup()
.findVirtual(this.getClass(), "onMessage", MethodType.methodType(void.class, String.class))
.bindTo(this);
}
}
}