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

Issue #2491 - FragmentExtension produces out of order frames 

+ Adding testcase to ensure no regression
+ All data frames that arrive, are now sent through the
  IteratingCallback to ensure that that input frame order
  is preserved.

Signed-off-by: Joakim Erdfelt <joakim.erdfelt@gmail.com>
This commit is contained in:
Joakim Erdfelt 2018-04-26 12:54:15 -05:00
parent 440d89750a
commit c596fcaac1
2 changed files with 121 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/extensions/fragment/FragmentExtension.java
View File

@ -60,19 +60,28 @@ public class FragmentExtension extends AbstractExtension
@Override
public void outgoingFrame(Frame frame, WriteCallback callback, BatchMode batchMode)
{
ByteBuffer payload = frame.getPayload();
int length = payload != null ? payload.remaining() : 0;
if (OpCode.isControlFrame(frame.getOpCode()) || maxLength <= 0 || length <= maxLength)
if (OpCode.isControlFrame(frame.getOpCode()))
{
// Skip fragment of control frames
nextOutgoingFrame(frame, callback, batchMode);
return;
}
FrameEntry entry = new FrameEntry(frame, callback, batchMode);
if (LOG.isDebugEnabled())
LOG.debug("Queuing {}", entry);
offerEntry(entry);
flusher.iterate();
// Handle everything else (data frames)
if(maxLength >= 1)
{
FrameEntry entry = new FrameEntry(frame, callback, batchMode);
if (LOG.isDebugEnabled())
LOG.debug("Queuing {}", entry);
offerEntry(entry);
flusher.iterate();
}
else
{
if (LOG.isDebugEnabled())
LOG.debug("Pass Through {}", frame);
nextOutgoingFrame(frame, callback, batchMode);
}
}
@Override
@ -124,14 +133,17 @@ public class FragmentExtension extends AbstractExtension
private boolean finished = true;
@Override
protected Action process() throws Exception
protected Action process()
{
if (finished)
{
current = pollEntry();
LOG.debug("Processing {}", current);
if (current == null)
{
LOG.debug("Processing IDLE", current);
return Action.IDLE;
}
LOG.debug("Processing {}", current);
fragment(current, true);
}
else
@ -146,9 +158,19 @@ public class FragmentExtension extends AbstractExtension
Frame frame = entry.frame;
ByteBuffer payload = frame.getPayload();
int remaining = payload.remaining();
int length = Math.min(remaining, maxLength);
finished = length == remaining;
if (first && finished)
{
// Simple send, no need to fragment.
if (LOG.isDebugEnabled())
LOG.debug("Skip Fragmentation {}", frame);
nextOutgoingFrame(frame, this, entry.batchMode);
return;
}
boolean continuation = frame.getType().isContinuation() || !first;
DataFrame fragment = new DataFrame(frame, continuation);
boolean fin = frame.isFin() && finished;
@ -186,7 +208,10 @@ public class FragmentExtension extends AbstractExtension
{
// Notify first then call succeeded(), otherwise
// write callbacks may be invoked out of order.
notifyCallbackSuccess(current.callback);
// only notify current (original) frame on completion
if (finished)
notifyCallbackSuccess(current.callback);
succeeded();
}

87
jetty-websocket/websocket-common/src/test/java/org/eclipse/jetty/websocket/common/extensions/FragmentExtensionTest.java
View File

@ -18,36 +18,48 @@
package org.eclipse.jetty.websocket.common.extensions;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.eclipse.jetty.io.ByteBufferPool;
import org.eclipse.jetty.io.MappedByteBufferPool;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;
import org.eclipse.jetty.websocket.api.BatchMode;
import org.eclipse.jetty.websocket.api.WebSocketPolicy;
import org.eclipse.jetty.websocket.api.extensions.ExtensionConfig;
import org.eclipse.jetty.websocket.api.extensions.Frame;
import org.eclipse.jetty.websocket.common.OpCode;
import org.eclipse.jetty.websocket.common.SaneFrameOrderingAssertion;
import org.eclipse.jetty.websocket.common.WebSocketFrame;
import org.eclipse.jetty.websocket.common.extensions.fragment.FragmentExtension;
import org.eclipse.jetty.websocket.common.frames.ContinuationFrame;
import org.eclipse.jetty.websocket.common.frames.PingFrame;
import org.eclipse.jetty.websocket.common.frames.TextFrame;
import org.eclipse.jetty.websocket.common.io.FutureWriteCallback;
import org.eclipse.jetty.websocket.common.test.ByteBufferAssert;
import org.eclipse.jetty.websocket.common.test.IncomingFramesCapture;
import org.eclipse.jetty.websocket.common.test.OutgoingFramesCapture;
import org.junit.Assert;
import org.junit.Test;
@SuppressWarnings("Duplicates")
public class FragmentExtensionTest
{
private static final Logger LOG = Log.getLogger(FragmentExtensionTest.class);
public ByteBufferPool bufferPool = new MappedByteBufferPool();
/**
@ -139,6 +151,7 @@ public class FragmentExtensionTest
/**
* Verify that outgoing text frames are fragmented by the maxLength configuration.
*
* @throws IOException on test failure
*/
@Test
@ -211,11 +224,12 @@ public class FragmentExtensionTest
}
/**
* Verify that outgoing text frames are fragmented by default configuration
* Verify that outgoing text frames are not fragmented by default configuration (which has no maxLength specified)
*
* @throws IOException on test failure
*/
@Test
public void testOutgoingFramesDefaultConfig() throws IOException
public void testOutgoingFramesDefaultConfig() throws Exception
{
OutgoingFramesCapture capture = new OutgoingFramesCapture();
@ -277,6 +291,7 @@ public class FragmentExtensionTest
/**
* Outgoing PING (Control Frame) should pass through extension unmodified
*
* @throws IOException on test failure
*/
@Test
@ -312,4 +327,72 @@ public class FragmentExtensionTest
Assert.assertThat("Frame.payloadLength", actual.getPayloadLength(), is(expected.remaining()));
ByteBufferAssert.assertEquals("Frame.payload", expected, actual.getPayload().slice());
}
/**
* Ensure that FragmentExtension honors the correct order of websocket frames.
*
* @see <a href="https://github.com/eclipse/jetty.project/issues/2491">eclipse/jetty.project#2491</a>
*/
@Test
public void testLargeSmallTextAlternating() throws Exception
{
final int largeMessageSize = 60000;
byte buf[] = new byte[largeMessageSize];
Arrays.fill(buf, (byte) 'x');
String largeMessage = new String(buf, UTF_8);
final int fragmentCount = 10;
final int fragmentLength = largeMessageSize / fragmentCount;
final int messageCount = 10000;
FragmentExtension ext = new FragmentExtension();
ext.setBufferPool(bufferPool);
ext.setPolicy(WebSocketPolicy.newServerPolicy());
ExtensionConfig config = ExtensionConfig.parse("fragment;maxLength=" + fragmentLength);
ext.setConfig(config);
SaneFrameOrderingAssertion saneFrameOrderingAssertion = new SaneFrameOrderingAssertion();
ext.setNextOutgoingFrames(saneFrameOrderingAssertion);
CompletableFuture<Integer> enqueuedFrameCountFut = new CompletableFuture<>();
CompletableFuture.runAsync(() -> {
// Run Server Task
int frameCount = 0;
BatchMode batchMode = BatchMode.OFF;
try
{
for (int i = 0; i < messageCount; i++)
{
int messageId = i;
FutureWriteCallback callback = new FutureWriteCallback();
WebSocketFrame frame;
if (i % 2 == 0)
{
frame = new TextFrame().setPayload(largeMessage);
frameCount += fragmentCount;
}
else
{
frame = new TextFrame().setPayload("Short Message: " + i);
frameCount++;
}
ext.outgoingFrame(frame, callback, batchMode);
callback.get();
}
enqueuedFrameCountFut.complete(frameCount);
}
catch (Throwable t)
{
enqueuedFrameCountFut.completeExceptionally(t);
}
});
int enqueuedFrameCount = enqueuedFrameCountFut.get(5, SECONDS);
int expectedFrameCount = (messageCount/2) * fragmentCount; // large messages
expectedFrameCount += (messageCount/2); // + short messages
assertThat("Saw expected frame count", saneFrameOrderingAssertion.frameCount, is(expectedFrameCount));
assertThat("Enqueued expected frame count", enqueuedFrameCount, is(expectedFrameCount));
}
}